• Introduction
• Two NETBLT Clients
• The NETBLT Implementation
  • The NETBLT Message
  • The NETBLT Endpoint
  • NETBLT Connection Management
  • Sending and Receiving
• The NETBLT Run-Time System
  • The Scheduler
  • The Message Handler
  • The Timer
  • The Control Sub-Message Sender
• Utilities and Miscellany
  • System Dependencies
  • Circular Lists
  • Bit Arrays
  • Stop Watches
  • The Keep List

This page last modified on 30 March 1998.

Two NETBLT Clients

send_file() sends the file named fname to the file-transfer server associated with UDP port port on the host with IP address host; the file is sent over the already opened (but not yet connected) NETBLT endpoint nbid.

<sndr.c>= [D->]
#include <stdio.h>
#include <unistd.h>
#include <fcntl.h>
#include <sys/types.h>
#include <sys/stat.h>
#include "netblt.h"

static void send_file(netblt_id nbid, u4b host, u2b port, char * fname) {

  u1b * bp;
  u4b i, f;
  struct stat b;

  <open the source file>

  <establish a NETBLT connection to the server>

  <send the file>

  close(f);
  free((char *) bp);
  }

Defines send_file (links are to index).

Open the source file and call fstat() to get its size.

<open the source file>= (<-U)

if ((f = open(fname, O_RDONLY)) == -1) {
  perror("open() failed");
  exit(1);
  }

if (fstat(f, &b) == -1) {
  perror("fstat() failed");
  exit(1);
  }

Connect to the file-transfer server using the file size as the transfer size; the other NETBLT parameters (e.g., buffer size and burst length) are left at their default values. If the connection is established, allocate a buffer based on the negotiated buffer size.

<establish a NETBLT connection to the server>= (<-U)

nbid->nnp.transfer_size = b.st_size;

connect_netblt(nbid, 's', host, port);
if (nbid->r_host == 0) {
  fprintf(stderr, "Connect attempt failed.\n");
  exit(1);
  }

bp = (u1b *) malloc(nbid->nnp.buffer_size);
assert(bp);

Send the file's contents to the file-transer server. The sender doesn't deal explicitly with buffer numbering or a final buffer containing less then the negotiated amount of data; NETBLT takes care of all that. To do so however, NETBLT must assume the sender is sending the file in start-to-end order.

<send the file>= (<-U)

for (i = nbid->buffers_per_transfer; i > 0; i--) {
  if (read(f, bp, nbid->nnp.buffer_size) == -1) {
    perror("read() failed");
    exit(1);
    }
  send_netblt(nbid, bp);
  }

The client of the pair is simple: it opens a NETBLT connection, sends a file over it, and closes the connection. open_netblt(0, 0) associates any host IP address and available UDP port with the opened NETBLT endpoint.

<sndr.c>+= [<-D]

void main(int argc, char * argv[]) {
  
  int r_port;
  char r_addr[20];
  extern u4b inet_addr(const char *);
  netblt_id nbid = open_netblt(0, 0);

  <process the sndr command line>

  send_file(nbid, str2ina(r_addr), r_port, argv[2]);

  close_netblt(nbid);

  exit(0);
  }

The file-transfer client command line indicates the IP address and UDP port associated with the file-transfer server, as well as the name of the file to send.

<process the sndr command line>= (<-U)

if (argc != 3) {
  fprintf(stderr, "Command format is \"%s host:port fname\".\n", argv[0]);
  exit(1);
  }

if (sscanf(argv[1], "%[^:]:%d", r_addr, &r_port) != 2) {
  fprintf(stderr, "missing host:port spec.\n");
  exit(1);
  }

<rcvr.c>= [D->]

#include "netblt.h"
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>

#ifndef nebtlt_ip_addr
#define netblt_ip_addr "0.0.0.0"
#endif

static void receive_file(netblt_id nbid) {
  
  u1b * bp;
  u4b j, ts;

  bp = (u1b *) malloc(nbid->nnp.buffer_size);
  assert(bp);

  ts = nbid->nnp.transfer_size;
  for (j = nbid->buffers_per_transfer; j > 0; j--) {
    fprintf(stderr, "receiving buffer %d\n", nbid->buffers_per_transfer-j+1);
    receive_netblt(nbid, bp);
    fwrite(bp, 1, minimum(nbid->nnp.buffer_size, ts), stdout);
    ts -= nbid->nnp.buffer_size;
    }

  free((char *) bp);
  }

Defines receive_file (links are to index).

The file-transfer server opens the NETBLT endpoint nbid at the IP address given by netblt_ip_addr with any available UDP port, prints a message (suitable for cutting and pasting) about what a client command line looks like, waits for a connection on the NETBLT endpoint, receives the file over the connection, closes the connection, and exits.

<rcvr.c>+= [<-D]

void main(void) {
  
  netblt_id nbid;
  u4b h;

  nbid = open_netblt(str2ina(netblt_ip_addr), 0);
  h = skt_addr(nbid->skt, 0);
  if (!h) h = host_addr(0);
  fprintf(stderr, "./sndr %s:%d\n", ina2str(h), port(nbid->skt));

  connect_netblt(nbid, 'r', 0, 0);

  receive_file(nbid);

  close_netblt(nbid);

  exit(0);
  }

The NETBLT Implementation

<NETBLT message definitions>= (U->) [D->]

#  define open_msg_type         0
#  define response_msg_type     1
#  define keepalive_msg_type    2
#  define quit_msg_type         3
#  define quitack_msg_type      4
#  define abort_msg_type        5
#  define data_msg_type         6
#  define ldata_msg_type        7
#  define nullack_msg_type      8
#  define control_msg_type      9
#  define refused_msg_type     10
#  define done_msg_type        11

#  define msg_type_count       12

The header found on the front of every NETBLT message; it's also the keepalive, quitack, and done message. The more interesting fields of the common header are:

• checksum. The usual 16-bit one's-compliment checksum over the message header.

• type. The msg_type values defined previously.

• length. In bytes; must be evenly divisible by four.

• local_port. At the message sender; the remote port at the message receiver.

• remote_port. At the message sender; the local port at the message receiver.

<NETBLT message definitions>+= (U->) [<-D->]

 typedef struct {
   u2b checksum;
   u1b version;
   u1b type;
   u2b length;
   u2b local_port;
   u2b remote_port;
   u2b padding;  
   } common_msg_hdr;

Defines common_msg_hdr (links are to index).

Some useful definitions related to the common message header. The maximum packet size is a little smaller than the Ethernet MTU.

<NETBLT message definitions>+= (U->) [<-D->]

typedef common_msg_hdr   msg;
typedef common_msg_hdr * msgp;

# define msg_length(_mp)      (((msgp) (_mp))->length)
# define msg_type(_mp)        (((msgp) (_mp))->type)
# define is_msg_type(_mp, _t) (msg_type(_mp) == _t ## _msg_type)

# define max_packet_size 1500

Defines is_msg_type, max_packet_size, msg, msg_length, msgp, msg_type (links are to index).

Message manipulation routines.

<NETBLT message definitions>+= (U->) [<-D->]

extern msgp
  make_msg(void);

extern void
  free_msg(msgp),
  reset_msg(msgp),
  send_msg(netblt_id, msgp),
  send_pkt(netblt_id, u4b, u2b, u1b *, u2b),
  set_checksum(msgp, bool),
  put_msg_data(msgp, const void *, u2b);

extern bool
  checksum_ok(msgp, bool);

An open or response message.

<NETBLT message definitions>+= (U->) [<-D->]

 typedef struct {
   common_msg_hdr cmh;
   u4b            connection_id;
   u4b            buffer_size;
   u4b            transfer_size;
   u2b            packet_size;
   u2b            burst_size;
   millisecs      burst_rate;
   seconds        death_timer_interval;
   u2b            reserved;
   u2b            max_concurrent_buffers;
   } or_msg, * or_msgp;

#define set_data_checksum_bit(_mp)   (((or_msgp) (_mp))->reserved |= 2)
#define clear_data_checksum_bit(_mp) (((or_msgp) (_mp))->reserved &= ~2)
#define get_data_checksum_bit(_mp)   (((or_msgp) (_mp))->reserved & 2)

Defines clear_data_checksum_bit, get_data_checksum_bit, or_msg, or_msgp, set_data_checksum_bit (links are to index).

An abort, control, quit, or refused message. This message is different from the common message header because it can take user data, while the common message header (in the form of the keepalive, quitack, or done message) does not.

<NETBLT message definitions>+= (U->) [<-D->]

typedef struct {
  common_msg_hdr cmh;
  } acqr_msg, * acqr_msgp;

Defines acqr_msg, acqr_msgp (links are to index).

An data or ldata message.

<NETBLT message definitions>+= (U->) [<-D->]

typedef struct {
  common_msg_hdr cmh;
  u4b            buffernumber;
  u2b            max_sequence;
  u2b            packet_number;
  u2b            data_checksum;
  u2b            reserved;
  } dl_msg, * dl_msgp;

#define set_ldata_bit(_mp) (((dl_msgp) (_mp))->reserved |= 1)
#define get_ldata_bit(_mp) (((dl_msgp) (_mp))->reserved & 1)

Defines dl_msg, dl_msgp, get_ldata_bit, set_ldata_bit (links are to index).

A nullack message.

<NETBLT message definitions>+= (U->) [<-D->]

typedef struct {
  common_msg_hdr cmh;
  u2b            max_sequence;
  u2b            burst_size;
  millisecs      burst_rate;
  u2b            padding;
  } n_msg, * n_msgp;

Defines n_msg, n_msgp (links are to index).

message_info is an keeps track of useful information about NETBLT messages; useful information includes a message's name, the number of bytes in a message's header, and whether or not a message can take user data. messag_inf is index by message type.

<NETBLT message definitions>+= (U->) [<-D->]

static const struct {
  char * name;
  u1b    header_size;
  bool   takes_data;
  } message_info[] = {
  {"open"     , sizeof(or_msg)  , true},
  {"response" , sizeof(or_msg)  , true},
  {"keepalive", sizeof(msg)     , false},
  {"quit"     , sizeof(acqr_msg), true},
  {"quitack"  , sizeof(msg)     , true},
  {"abort"    , sizeof(acqr_msg), true},
  {"data"     , sizeof(dl_msg)  , true},
  {"ldata"    , sizeof(dl_msg)  , true},
  {"nullack"  , sizeof(n_msg)   , false},
  {"control"  , sizeof(acqr_msg), true},
  {"refused"  , sizeof(acqr_msg), true},
  {"done"     , sizeof(msg)     , false}
  };                                                                 

#define msg_header_size(_mp) (message_info[msg_type(_mp)].header_size)

#define msg_takes_data(_mp)  (message_info[msg_type(_mp)].takes_data)

#define msg_name(_mp)        (message_info[msg_type(_mp)].name)

#define msg_data(_mp)        (((u1b *) (_mp)) + msg_header_size(_mp))

Defines message_info, msg_data, msg_header_size, msg_name, msg_takes_data (links are to index).

Control submessages are stored within control messages. These are the types of control submessages.

<NETBLT message definitions>+= (U->) [<-D->]

#  define go_submessage_type     0
#  define ok_submessage_type     1
#  define resend_submessage_type 2

The common header for control submessages, and the go control submessage.

<NETBLT message definitions>+= (U->) [<-D->]

typedef struct {
  u1b           type;
  u1b           padding;
  u2b           sequence_number;
  buffer_number buffernumber;
  } common_submessage_header,
    go_msg, * go_msgp;

typedef common_submessage_header * submsgp;

#define submsg_type(_smp)       (((submsgp) (_smp))->type)
#define is_submsg_type(_mp, _t) (submsg_type(_mp) == _t ## _submessage_type)
#define sm_buffer_number(_mp)   (((submsgp) (_mp))->buffernumber)

Defines common_submessage_header, go_msg, go_msgp, is_submsg_type, sm_buffer_number, submsgp, submsg_type (links are to index).

The ok control submessage.

<NETBLT message definitions>+= (U->) [<-D->]

typedef struct {
  common_submessage_header csh;
  u2b                      burst_size;
  millisecs                burst_rate;
  millisecs                control_timer;
  u2b                      padding;
  } ok_msg, * ok_msgp;

Defines ok_msg, ok_msgp (links are to index).

The resend control submessage.

<NETBLT message definitions>+= (U->) [<-D->]

typedef struct {
  common_submessage_header csh;
  u2b                      pkt_cnt;
  u2b                      padding;
  } resend_msg, * resend_msgp;

#define resend_pkt_count(_m) (((resend_msgp) (_m))->pkt_cnt)

#define resend_pkt_start(_m) ((u2b *) (((u1b *) (_m)) + sizeof(resend_msg)))

Defines resend_msg, resend_msgp, resend_pkt_count, resend_pkt_start (links are to index).

Some routines associated with control submessages.

<NETBLT message definitions>+= (U->) [<-D]

extern void
  control_msg_iter(acqr_msgp cmp, bool (* f)(submsgp, void *), void *),
  dump_submessage(FILE *, submsgp),
  dump_message(FILE *, msgp);

extern u2b
  smsg_length(submsgp);

extern bool
  has_control_msg(msgp, int, buffer_number);

The dot h file.

<msg.h>=

#ifndef _msg_h_defined_
#define _msg_h_defined_

#include "defs.h"

<NETBLT message definitions>

#endif

Defines "msg.h" (links are to index).

make_msg() returns a pointer to a new NETBLT message, which was created by getting one off either the free list or the heap.

<message code>= (U->) [D->]

msgp make_msg(void) {
  msgp mp;

  if (m_free_list) {
    mp = m_free_list;
    m_free_list = *((msgp *) mp);
    }
  else {
    if (!kl) kl = init_kl();
    mp = (msgp) malloc(max_packet_size);
    assert(mp);
    }
  reset_msg(mp);
  add_kl(kl, mp);

  return mp;
  }

Defines make_msg (links are to index).

free_msg() frees message pointed to by mp by hanging it on the free list.

<message code>+= (U->) [<-D->]

void free_msg(msgp mp) {
  drop_kl(kl, mp);
  *((msgp *) mp) = m_free_list;
  m_free_list = mp;
  }

Defines free_msg (links are to index).

reset_msg() sets the message pointed to by mp to an initial state.

<message code>+= (U->) [<-D->]

void reset_msg(msgp mp) {
  msg_length(mp) = 0;
  mp->version = netblt_version;
  }

Defines reset_msg (links are to index).

send_msg() sends the message pointed to by mp over the NETBLT endpoint nbid.

<message code>+= (U->) [<-D->]

void send_msg(netblt_id nbid, msgp mp) {

  mp->version = netblt_version;
  mp->remote_port = nbid->r_port;
  mp->local_port = port(nbid->skt);

  <set the message length>

  set_checksum(mp, nbid->nnp.checksum_data);

  send_pkt(nbid, nbid->r_host, nbid->r_port, (u1b *) mp, msg_length(mp));
  }

Defines send_msg (links are to index).

If the message has non-zero length, assume it's been set elsewhere (in particular, assume it's been set by the routine that stores user data in a message). Otherwise, there's no user data and the message size is just the header size.

<set the message length>= (U->)

if (msg_length(mp) == 0)
  msg_length(mp) = msg_header_size(mp);

if (msg_length(mp) & 3) {
  fprintf(stderr, "%s msg has length = %d.\n", msg_name(mp), msg_length(mp));
  exit(1);
  }

Send the message pkt of size pkt_size bytes over the NETBLT endpoint nbid to the corresponding NETBLT endpoint on host host with UDP port port.

<message code>+= (U->) [<-D]

void send_pkt(netblt_id nbid, u4b host, u2b port, u1b * pkt, u2b pkt_size) {

  int e;
  struct sockaddr_in sa;

  assert(pkt_size > 0);

  bzero((char *) &sa, sizeof(sa));
  sa.sin_family = AF_INET;
  sa.sin_addr.s_addr = htonl(host);
  sa.sin_port = htons(port);

  debugp((stderr, "Send %s to %s:%d.\n", msg_name(name), ina2str(host), port));

  e = sendto(nbid->skt, (char *) pkt, pkt_size, 0,
             (struct sockaddr *) &sa, sizeof(sa));
  check(e < 0, "sendto()");
  }

Defines send_pkt (links are to index).

The checksum is the usual one's-compliment arithmatic over 16-bit words.

<message checksum code>= (U->) [D->]

static u2b checksum(u2b * bp, u2b size) {
  u2b cs = 0;
  u2b i;

  for (i = 0; i < size; i++) cs ^= *bp++;
  return ~cs;
  }

Defines checksum (links are to index).

Specialize checksum() to compute message header and message data checksums.

<message checksum code>+= (U->) [<-D->]

#define hdr_checksum(_mp) \
  checksum((u2b *) (_mp), msg_header_size(_mp)/2)

#define data_checksum(_mp) \
  checksum(((u2b *) (_mp)) + msg_header_size(_mp)/2, \
           (msg_length(_mp) - msg_header_size(_mp)/2))

Defines data_checksum, hdr_checksum (links are to index).

Given mp, a pointer to a message, compute the message's header checksum. If data_cs is true and the message is a data or ldata message, also compute the data checksum.

<message checksum code>+= (U->) [<-D->]

void set_checksum(msgp mp, bool data_cs) {

  if (is_msg_type(mp, data) || is_msg_type(mp, ldata)){
    ((dl_msgp) mp)->data_checksum = 0;
    if (data_cs) ((dl_msgp) mp)->data_checksum = data_checksum(mp);
    }

  mp->checksum = 0;
  mp->checksum = hdr_checksum(mp);
  }

Defines set_checksum (links are to index).

Given mp, a pointer to a message, return true if and only if the message's header checksum is correct. If data_cs is true and the message is a data or ldata message, return true if and only if both the header and data checksums are correct.

<message checksum code>+= (U->) [<-D]

bool checksum_ok(msgp mp, bool data_cs) {
  u2b cs = mp->checksum;
  mp->checksum = 0;
  if (cs != hdr_checksum(mp)) {
    fprintf(stderr, "checksum failure: %x != %x.\n", cs , hdr_checksum(mp));
    return false;
    }
  if ((is_msg_type(mp, data) || is_msg_type(mp, ldata)) && data_cs)
    if (((dl_msgp) mp)->data_checksum != data_checksum(mp)) return false;
  return true;
  }

Defines checksum_ok (links are to index).

Given mp, a pointer to a message, and data, a pointer to data_size bytes of data, put_msg_data() stores the data in the message, setting the messages length field.

<miscellaneous message code>= (U->) [D->]

void put_msg_data(msgp mp, const void * data, u2b data_size) {

  char * cp;

  assert(msg_takes_data(mp));
  assert(msg_length(mp) == 0);
  assert(data_size > 0);

  cp = (char *) msg_data(mp);

  bzero(cp + ((data_size - 1) & 0xfffc), 4);
  byte_copy((char *) cp, (const char *) data, data_size);
  msg_length(mp) =
    msg_header_size(mp) + data_size + ((4 - (data_size & 0x03)) & 0x03);
  }

Defines put_msg_data (links are to index).

Given smp, a pointer to a control sub-message, dump_submessage() writes a description of smp to the file out.

<miscellaneous message code>+= (U->) [<-D->]

void dump_submessage(FILE * out, submsgp smp) {

  fprintf(out, "type = %d (%s); sequence number = %d; buffer number = %ld\n",
          submsg_type(smp), 
          (!submsg_type(smp) ? "go" :
           (submsg_type(smp) == 1 ? "ok" : "resend")),
          smp->sequence_number, sm_buffer_number(smp));

  switch (submsg_type(smp)) {
    case go_submessage_type:
      break;
      
    case ok_submessage_type: {
      ok_msgp omp = (ok_msgp) smp;
      fprintf(out, "burst rate = %d; burst size = %d; control timer = %d\n",
              omp->burst_rate, omp->burst_size, omp->control_timer);
      break;
      }

    case resend_submessage_type: {
      u2b * pnop;
      u1b i;
      resend_msgp rmp = (resend_msgp) smp;
      fprintf(out, "missing packet count = %d\nmissing packets: ",
              rmp->pkt_cnt);

      pnop = resend_pkt_start(rmp);
      for (i = 0; i < rmp->pkt_cnt; i++) fprintf(out, " %d", *pnop++);
      fprintf(out, "\n");
      break;
      }
    default:
      fprintf(stderr, "Unrecognized sub-message %d in dump_submessage().\n",
              submsg_type(smp));
      exit(1);
    }
  }

Defines dump_submessage (links are to index).

Given mp, a pointer to a message, dump_message() writes a description of mp to the file out.

<miscellaneous message code>+= (U->) [<-D->]

void dump_message(FILE * out, msgp mp) {

  fprintf(out, "type = %d (%s); version = %d; length = %d\n",
          msg_type(mp), msg_name(mp), mp->version, msg_length(mp));
  fprintf(out, "local port = %d; remote port = %d\n",
          mp->local_port, mp->remote_port);
  }

Defines dump_message (links are to index).

Given mp, a pointer to a control sub-message, smsg_length() returns the size of *mp in bytes.

<miscellaneous message code>+= (U->) [<-D->]

u2b smsg_length(submsgp mp) {
  switch (submsg_type(mp)) {
    case go_submessage_type: return sizeof(go_msg);
    case ok_submessage_type: return sizeof(ok_msg);
    case resend_submessage_type:
      return sizeof(resend_msg) +
             2*(((resend_msgp) mp)->pkt_cnt + (((resend_msgp) mp)->pkt_cnt & 1));
    default:
      assert(!"unknown sub-message type in smsg_length()");
    }
  return 0;
  }

Defines smsg_length (links are to index).

has_control_msg() returns true iff the message pointed to by mp contains a sub-message of type cmt for buffer bno.

<miscellaneous message code>+= (U->) [<-D->]

bool has_control_msg(msgp mp, int cmt, buffer_number bno) {

  u1b * p;
  const u1b * end = ((u1b *) mp) + msg_length(mp);
  
  if (!is_msg_type(mp, control)) return false;

  for (p = msg_data(mp); p < end; p += smsg_length((submsgp) p))
    if ((submsg_type(p) == cmt) && (sm_buffer_number(p) == bno)) return true;

  return false;
  }

Defines has_control_msg (links are to index).

control_msg_iter() runs through the control message pointed to by cmp and calls f for each control sub-message found, passing in a pointer to the control sub-message and the otherwise uninterpreted (by control_msg_iter()) argument a.

<miscellaneous message code>+= (U->) [<-D]

void control_msg_iter(acqr_msgp cmp, bool (* f)(submsgp, void *), void * a) {

  u1b * p = msg_data(cmp);
  const u1b * end = p + msg_length(cmp);
  
  while ((p < end) && (f)((submsgp) p, a)) p += smsg_length((submsgp) p);
  }

Defines control_msg_iter (links are to index).

<msg.c>=

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "msg.h"
#include "keep-list.h"
#include "netblt.h"


/* Sundries. */

   static msgp m_free_list = 0;
   static keep_list kl;

<message code>

<message checksum code>

<miscellaneous message code>

The NETBLT Endpoint

• buffers_per_transfer. The total number of buffers that will be used in the transfer.

• current_bno. The number of the buffer being sent or expected next; in the case of multi-buffer transfers, it's the smallest buffer number in the sequence.

• last_payload_size. The amount of data stored in the last packet sent in the transfer.

• next_seq_no. On the sending side of the connection, it's the next sequence number to send; on the receiving side of the connection, it's the next sequence number to expect.

• nnp. The paramater values negotiated by both ends of the connection.

• nullack_mhid, death_mhid, heartbeat_mhid. The ids for the vita that acknowledge null messages, check for death time-outs, and issue keepalive messages, respectively.

• packets_per_buffer. The number of packets needed to transfer a full buffer.

• payload_size. The amount of data transfered in a full packet (this differs from the packet size due to header overhead).

• r_host. The IP address of the other (remote) end of the connection.

• r_port. The UDP port number of the other (remote) end of the connection.

• skt. The UDP socket used to send and receive data.

<The NETBLT endpoint data structure>= (U->)

typedef struct Netblt_id {
  int                          skt;
  u4b                          r_host;
  u2b                          r_port;
  netblt_negotiated_parameters nnp;
  u2b                          buffers_per_transfer;
  u2b                          packets_per_buffer;
  u2b                          payload_size;
  u2b                          last_payload_size;
  buffer_number                current_bno;
  sequence_number              next_seq_no;
  mhandler_id                  nullack_mhid, death_mhid, heartbeat_mhid;
  } Netblt_id;

Defines Netblt_id (links are to index).

The NETBLT protocol lets both sides of a connection negotiate values for various parameters affecting the data transfer between the sides. The parameters available for negotiation are

• buffer_size. The buffer size in bytes.

• burst_interval. The length of a packet-sending burst, in milliseconds.

• checksum_data. Indicate if data being send should be checksummed.

• death_timer_interval. How long the other side is going to wait without getting a message before it declares this side to be dead. Not a negotiated value; this side uses this value to set its own heartbeat timer to insure the other side won't assume an untimley death.

• max_concurrent buffers. The number of buffers being sent during a single transfer.

• packet_size. The packet size in bytes.

• packets_per_burst. The number of packets sent in a packet-sending burst.

• transfer_size. The transfer size in bytes. It's not really negotiable, the receiver can either accept or reject.

<the NETBLT negotiated parameters>= (U->)

typedef struct {
  u4b        transfer_size;
  u4b        buffer_size;
  u2b        packet_size;
  u2b        packets_per_burst;
  millisecs  burst_interval;
  u2b        death_timer_interval;
  bool       checksum_data;
  u2b        max_concurrent_buffers;
  } netblt_negotiated_parameters, * netblt_negotiated_parametersp;

Defines netblt_negotiated_parameters, netblt_negotiated_parametersp (links are to index).

open_netblt() creates a NETBLT endpoint; close_netblt() deletes a NETBLT endpoint.

<netblt.h>=

#include "m-handler.h"

#ifndef _netblt_h_defined_
#define _netblt_h_defined_

#define netblt_version 1

<the NETBLT negotiated parameters>

<The NETBLT endpoint data structure>

extern netblt_id open_netblt(u4b, u2b);
extern void close_netblt(netblt_id);

#include "connect.h"
#include "sender.h"
#include "receiver.h"

#endif

Defines "netblt.h" (links are to index).

Creating a NETBLT endpoint assigns a UDP port to it and initializes the internal parameters.

<netblt.c>= [D->]

#include <sys/types.h>
#include <sys/socket.h>
#include "netblt.h"
#include "msg.h"
#include "scheduler.h"
#include "system.h"

netblt_id open_netblt(u4b host, u2b port) {

  netblt_id nbid = (netblt_id) malloc(sizeof(Netblt_id));
  extern int socket(int, int, int);

  assert(nbid);

  nbid->skt = socket(PF_INET, SOCK_DGRAM, 0);
  check(nbid->skt < 0, "socket()");

  bind_skt(nbid->skt, host, port);

  nbid->r_host = nbid->r_port = 0;

  <set default values for negotiated parameters>

  nbid->nullack_mhid = nbid->death_mhid = nbid->heartbeat_mhid = 0;
  nbid->current_bno = 0;
  nbid->next_seq_no = 1;

  init_scheduler(nbid->skt);

  return nbid;
  }

Defines open_netblt (links are to index).

The only value the NETBLT client must set is transfer_size. The burst values give a transfer rate of 1400 byt/pkt*((5 pkt/bst)/(45 msec/bst))*(1000 msec/sec) = 1.2 mbit/sec

<set default values for negotiated parameters>= (<-U)

nbid->nnp.transfer_size = 0;
nbid->nnp.buffer_size = 100000;
nbid->nnp.packet_size = 1400;           /* must be zero mod 4 */
nbid->nnp.packets_per_burst = 5;
nbid->nnp.burst_interval = 45;
nbid->nnp.death_timer_interval = 30;
nbid->nnp.checksum_data = false;
nbid->nnp.max_concurrent_buffers = 1;

To delete a NETBLT endpoint, cancel the null-ack message handler if it exists, and close the UDP socket associated with the endpoint.

<netblt.c>+= [<-D->]

void close_netblt(netblt_id nbid) {

  if (nbid->nullack_mhid) cancel_mhandler(nbid->nullack_mhid);
  term_scheduler();

  check(close(nbid->skt), "close()");
  free((char *) nbid);

  }

Defines close_netblt (links are to index).

If all packets in a buffer have been received, the receiving end of a NETBLT connction will respond with a null-ack message to update the sending side's sequence numbers. The null-ack message will also contain the new values for burst-interval length and packets per burst computed by the receiving side during the last buffer transfer.

The sending-side nullack_handler() deals with null-ack messages from the receiver.

<netblt.c>+= [<-D]

void nullack_handler(void * ap, void * iap) {
  
  netblt_id nbid = (netblt_id) ap;
  n_msgp nmp = (n_msgp) (((msg_bundlep) iap)->mp);

  nbid->nnp.packets_per_burst =
    minimum(nbid->nnp.packets_per_burst, nmp->burst_size);
  nbid->nnp.burst_interval =
    minimum(nbid->nnp.burst_interval, nmp->burst_rate);
  }

Defines nullack_handler (links are to index).

NETBLT Connection Management

<connect.h>=

#ifndef _connect_h_defined_
#define _connect_h_defined_

#include "netblt.h"

extern void connect_netblt(netblt_id, char, u4b, u2b);

#endif

Defines "connect.h" (links are to index).

connect_netblt() accepts the unconnected NETBLT endpoint nbid and an indication if this endpoint is on the 's'ending or 'r'eceiving end. If nbid is the sending endpoint, host and port are the IP address and UDP port of the receiving end; if nbid is the receiving endpoint, host and port are the IP address and UDP port of an acceptable sending end.

The work of making the connections falls to the routines sconnect() and rconnect(); connect_netblt() itself just initializes some read-only parameters in the NETBLT endpoint.

<connect_netblt()>= (U->)

void connect_netblt(netblt_id nbid, char what, u4b host, u2b port) {

  (what == 'r' ? rconnect : sconnect)(nbid, host, port);

  nbid->payload_size = nbid->nnp.packet_size - sizeof(dl_msg);
  nbid->buffers_per_transfer =
    ceiling(nbid->nnp.transfer_size, nbid->nnp.buffer_size);
  nbid->packets_per_buffer =
    ceiling(nbid->nnp.buffer_size, nbid->payload_size);
  nbid->last_payload_size = nbid->nnp.buffer_size - 
    ((nbid->packets_per_buffer - 1)*(nbid->payload_size));
  }

Defines connect_netblt (links are to index).

rconnect() attempts to establish nbid as the receiving-end of a NETBLT connection; host and port give the IP address and UDP port number of acceptable peers, or are 0 if any peer is acceptable.

On the receiving side of a connection, rconnect() schedules a handler for connection requests, and then waits for the handler to terminate. At termination, rconnect() checks to see if a connection has been established and, if it has, schedules a handler for control messages back to the sender and a handler for null acknowledgement messages.

<rconnect()>= (U->)

static void rconnect(netblt_id nbid, u4b host, u2b port) {

  mhandler_id mhid;
  extern void nullack_handler(void *, void *);

  ca.nbid = nbid;
  ca.host = host;
  ca.port = port;
  ca.done = false;

  mhid = schedule_mhandler(set_msgtype(0, open), receive_open, &ca);
  run_scheduler(rtf);
  cancel_mhandler(mhid);

  if (nbid->r_host) {
    init_ctlsdr(nbid);
    nbid->nullack_mhid = schedule_mhandler(set_msgtype(0, nullack),
                                           nullack_handler, nbid);
    }
  }

Defines rconnect (links are to index).

sconnect() attempts to establish nbid as the sending-end of a NETBLT connection; the receiving end is supposed to reside at the UDP port prt on the host with IP address host.

<sconnect()>= (U->)

static void sconnect(netblt_id nbid, u4b host, u2b prt) {

  mhandler_id mhid;
  const u1b z[1] = {0};
  msgp mp;

  <sconnect() code>
  }

Defines sconnect (links are to index).

First construct an NETBLT open message to send to the other end; the message sends no user data.

<sconnect() code>= (U->) [D->]

sca.omp = (or_msgp) make_msg();
msg_type(sca.omp) = open_msg_type;

sca.omp->transfer_size = nbid->nnp.transfer_size;
sca.omp->buffer_size = nbid->nnp.buffer_size;
sca.omp->packet_size = nbid->nnp.packet_size;
sca.omp->burst_size = nbid->nnp.packets_per_burst;
sca.omp->burst_rate = nbid->nnp.burst_interval;
sca.omp->max_concurrent_buffers = nbid->nnp.max_concurrent_buffers;
sca.omp->death_timer_interval = nbid->nnp.death_timer_interval;
sca.omp->reserved = 0;
if (nbid->nnp.checksum_data) set_data_checksum_bit(sca.omp);

mp = (msgp) sca.omp;
mp->version = netblt_version;
mp->remote_port = prt;
mp->local_port = port(nbid->skt);
put_msg_data(mp, z, 1);
set_checksum(mp, nbid->nnp.checksum_data);

When the open message is ready, sconnect() schedules two handlers: receive_response() deals with the response received from the other end of the connection, and send_open() is responsible for sending the open message to the other end.

<sconnect() code>+= (U->) [<-D]

sca.retries_left = 10;
sca.response_timeout = 1000;
sca.nbid = nbid;
sca.r_host = host;
sca.r_port = prt;
sca.done = false;

mhid = schedule_mhandler(set_msgtype(set_msgtype(0, response), refused),
                         receive_response, &sca);
send_open(&sca, 0);

run_scheduler(stf);

cancel_mhandler(mhid);

free_msg((msgp) sca.omp);

The send_open() handler is responsible for sending an open message. NETBLT allows for time-outs and a fixed number of retries on opens; if send_open() has retries left, it send the open message and schedules itself to send it again at the end of the next time-out interval. If there are no retries left, send_open() terminates itself.

<send_open()>= (U->)

static void send_open(void * ap, void * iap) {

  sconnect_argsp scap = (sconnect_argsp) ap;

  if (scap->retries_left == 0) {
    scap->done = true;
    return;
    }

  send_pkt(scap->nbid, scap->r_host, scap->r_port,
           (u1b *) scap->omp, msg_length(scap->omp));

  scap->tid = schedule_timer(scap->response_timeout*1000, send_open, ap);
  scap->retries_left--;
  }

Defines send_open (links are to index).

receive_response() handles the message pointed to by iap using the state pointed to by iap.

<receive_response()>= (U->)

static void receive_response(void * ap, void * iap) {

  sconnect_argsp scap = (sconnect_argsp) ap;
  msg_bundlep mbp = (msg_bundlep) iap;
  or_msgp rmp = (or_msgp) mbp->mp;

  <receive_response() code>
  }

Defines receive_response (links are to index).

If the message isn't from the expected peer, ignore it.

<receive_response() code>= (U->) [D->]

if ((mbp->host != scap->r_host) || (mbp->mp->local_port != scap->r_port))
  return;

If the message is a response, tuck away the negotiated parameters; otherwise, the message better be a rejection.

<receive_response() code>+= (U->) [<-D->]

if (is_msg_type(rmp, response)) {
  assert(scap->nbid->nnp.transfer_size == rmp->transfer_size);

  scap->nbid->nnp.buffer_size = rmp->buffer_size;
  scap->nbid->nnp.packet_size = rmp->packet_size;
  scap->nbid->nnp.packets_per_burst = rmp->burst_size;
  scap->nbid->nnp.burst_interval = rmp->burst_rate;
  scap->nbid->nnp.max_concurrent_buffers = rmp->max_concurrent_buffers;
  scap->nbid->nnp.death_timer_interval = rmp->death_timer_interval;
  scap->nbid->nnp.checksum_data = get_data_checksum_bit(rmp);
  scap->nbid->r_host = scap->r_host;
  scap->nbid->r_port = scap->r_port;
  }
else
  assert(is_msg_type(rmp, refused));

In either case, receive_response() cancels itself and indicates it's done.

<receive_response() code>+= (U->) [<-D]

cancel_timer(scap->tid);
scap->done = true;

The receive_open() procedure accepts the connection information pointed to by ap and the open message pointed to by ip and returns an accept message to the sender. The parameter negotiation consists of setting each para meter to the lower of the two acceptable values.

<receive_open()>= (U->)

static void receive_open(void * ap, void * ip) {

  connect_argsp cap = (connect_argsp) ap;
  msg_bundlep mbp = (msg_bundlep) ip;
  or_msgp omp = (or_msgp) mbp->mp;
  u2b dti = cap->nbid->nnp.death_timer_interval;

  <negotiate transfer parameters>

  msg_type(omp) = response_msg_type;
  msg_length(omp) = 0;

  cap->nbid->r_port = ((msgp) omp)->local_port;
  cap->nbid->r_host = mbp->host;
  send_msg(cap->nbid, (msgp) omp);

  cap->done = true;
  }

Defines receive_open (links are to index).

Negotation just takes the smaller of the values expected by this endpoint and offered by the other endpoint.

<negotiate transfer prameters>=

cap->nbid->nnp.transfer_size = omp->transfer_size;
omp->buffer_size = cap->nbid->nnp.buffer_size =
  minimum(cap->nbid->nnp.buffer_size, omp->buffer_size);
omp->packet_size = cap->nbid->nnp.packet_size =
  minimum(cap->nbid->nnp.packet_size, omp->packet_size);
omp->burst_size = cap->nbid->nnp.packets_per_burst =
  minimum(cap->nbid->nnp.packets_per_burst, omp->burst_size);
omp->burst_rate = cap->nbid->nnp.burst_interval =
  minimum(cap->nbid->nnp.burst_interval, omp->burst_rate);
omp->max_concurrent_buffers = cap->nbid->nnp.max_concurrent_buffers =
  minimum(cap->nbid->nnp.max_concurrent_buffers,omp->max_concurrent_buffers);
cap->nbid->nnp.death_timer_interval = omp->death_timer_interval;
omp->death_timer_interval = dti;
cap->nbid->nnp.checksum_data =
  cap->nbid->nnp.checksum_data || get_data_checksum_bit(omp);
if (cap->nbid->nnp.checksum_data) set_data_checksum_bit(omp);

To operate properly, the connection routines need to know

• done True if and only if connection management has been completed (successfully or otherwise).

• host The local host's IP address.

• nbid A pointer to the end-point on whose behalf this connection is being made.

• port The end-point's associated UDP port.

<connection state>= (U->) [D->]

typedef struct {
  netblt_id   nbid;
  u4b         host;
  u2b         port;
  bool        done;
  } connect_args, * connect_argsp;

Defines connect_args, connect_argsp (links are to index).

The connection manager on the sending side needs to know a bit more than does the connection manager on the receiving side.

• done True if and only if connection management has been completed (successfully or otherwise).

• nbid A pointer to the NETBLT end-point on whose behalf this connection is being made.

• omp A pointer to the open message being sent.

• r_host The IP address associated with the receiving NETBLT end-point.

• r_port The UDP port associated with the receiving NETBLT end-point.

• response_timeout The number of milliseconds that should elapse without receiving a response from the other side before the open message is retransmitted.

• retries_left The number of times the open message should be sent.

• tid The retransmission timer's id.

<connection state>+= (U->) [<-D]

 typedef struct {
   or_msgp     omp;
   u1b         retries_left;
   millisecs   response_timeout;
   bool        done;
   timer_id    tid;
   u4b   r_host;
   u2b   r_port;
   netblt_id   nbid;
   } sconnect_args, * sconnect_argsp;

Defines sconnect_args, sconnect_argsp (links are to index).

The scheduler terimation functions tell the scheduler when it should return to the sending end (stf) or receiving end (rtf).

<scheduler termination functions>= (U->)

static bool rtf(void) {
  return ca.done;
  }

static bool stf(void) {
  return sca.done;
  }

Defines rtf, stf (links are to index).

<connect.c>=

#include "connect.h"
#include "m-handler.h"
#include "timer.h"
#include "scheduler.h"
#include "control-sender.h"

<connection state>

/* Sundries. */

   static connect_args ca;
   static sconnect_args sca;

<scheduler termination functions>

<receive_open()>

<receive_response()>

<send_open()>

<rconnect()>

<sconnect()>

<connect_netblt()>

Sending and Receiving

<sender.h>=
#ifndef _sender_h_defined_
#define _sender_h_defined_

#include "netblt.h"

extern void send_netblt(netblt_id, u1b *);

#endif

Defines "sender.h" (links are to index).

The data needed to resend packets are:

• missing_pkt_cnt: the number of missing packets.

• missing_pkts: an array of packet numbers; packet p is missing if and only if missing_pkts[i] = p for some 0 <= i < missing_pkt_cnt.

<missing packet data>= (U->)

   typedef struct {
     u2b * missing_pkts;
     u2b   missing_pkt_cnt;
     } resend_data, * resend_datap;

Defines resend_data, resend_datap (links are to index).

<sender.c>= [D->]

#include "sender.h"
#include "m-handler.h"
#include "timer.h"
#include "scheduler.h"
#include "msg.h"

/* What the buffer sending routines need to know. */

   <missing packet data>

   typedef struct {
     u1b           * buffer;
     buffer_number   bno;
     u2b             packets_remaining;
     bool            done;
     netblt_id       nbid;
     mhandler_id     go_mhid, ok_mhid, resend_mhid, done_mhid;
     u2b             reserved;
     millisecs       control_ack_timeout;
     bool            last_buffer;
   } sb_args, * sb_argsp;


/* Sundries. */

   static sb_args sba;
   static void get_resend(void *, void *), get_ok(void *, void *);

#  define has_controlm(_m, _t, _a) \
     has_ctl_msg((acqr_msgp) (_m), _t ## _submessage_type, _a)

   static bool do_ok_msg(msgp, sb_argsp);
   static void get_done(void *, void *);

Defines has_controlm, sb_args, sb_argsp (links are to index).

Given cmp, a pointer to a control message, mt, a sub-message type, and sbap, a pointer to sending buffer information, had_ctl_msg() returns a zero if cmp does not contain a proper sub-message of type mt; otherwise it returns a pointer to the sub-message.

A sub-message is proper if it contains the expected next sequence number and it refers to the buffer currently being sent. Note this code doesn't optimize things by accepting sub-messages with sequence numbers ahead of the expected next sequence number.

<sender.c>+= [<-D->]

static submsgp has_ctl_msg(acqr_msgp cmp, int mt, sb_argsp sbap) {

  u1b * p;
  const u1b * end = ((u1b *) cmp) + msg_length(cmp);

  for (p = msg_data(cmp); p < end; p += smsg_length((submsgp) p)) {
    submsgp smp = (submsgp) p;
    if ((smp->sequence_number == sbap->nbid->next_seq_no) &&
        (sm_buffer_number(smp) == sbap->bno) &&
        (smp->type == mt)) {
      sbap->nbid->next_seq_no++;
      return smp;
      }
    }
  return 0;
  }

Defines has_ctl_msg (links are to index).

<sender.c>+= [<-D->]

static void get_done(void * ap, void * iap) {

  sb_argsp sbap = (sb_argsp) ap;
  msgp mp = (msgp) (((msg_bundlep) iap)->mp);

  if (do_ok_msg(mp, sbap)) return;

  if (is_msg_type(mp, done)) {
    sbap->done = true;
    return;
    }

  assert(!"unexpected message type in get_done");
  }

Defines get_done (links are to index).

Given mp, a pointer to a message, and sbap, a pointer to sending-buffer information, do_ok_msg() returns true if the message pointed to by mp contains an ok sub-message. In addition, if do_ok_msg() finds a valid ok sub-message, it will send a null-acknowledgement message back to the receiving end and, if necessary, schedule a handler for a done message.

<sender.c>+= [<-D->]

static bool do_ok_msg(msgp mp, sb_argsp sbap) {

  ok_msgp omp;
  n_msg nm;

  <do_ok_msg() code>

  return true;
  }

Defines do_ok_msg (links are to index).

If the message isn't a control message, or the control message doesn't contain a valid ok sub-message, return false.

<do_ok_msg() code>= (<-U) [D->]

if (!is_msg_type(mp, control)) return false;

omp = (ok_msgp) has_controlm(mp, ok, sbap);
if (!omp) return false;

If the current sending buffer is the last one, schedule a message handler to receive the done message from the receiving side.

<do_ok_msg() code>+= (<-U) [<-D->]

if (sbap->last_buffer)
  sbap->done_mhid = 
    schedule_mhandler(set_msgtype(set_msgtype(0, control), done),
                      get_done, sbap);

Acknowledge the ok sub-message. The new negotiated values are accepted as is.

<do_ok_msg() code>+= (<-U) [<-D]

msg_type(&nm) = nullack_msg_type;
msg_length(&nm) = 0;
nm.max_sequence = sbap->nbid->next_seq_no - 1;
sbap->nbid->nnp.packets_per_burst = nm.burst_size = omp->burst_size;
sbap->nbid->nnp.burst_interval = nm.burst_rate = omp->burst_rate;
sbap->control_ack_timeout = omp->control_timer;
send_msg(sbap->nbid, (msgp) &nm);

The procedure send_pkts() sends the next burst of packets; ap is a pointer to the sending-buffer information (because send_pkts() reacts to timer events, iap is 0).

<sender.c>+= [<-D->]

static void send_pkts(void * ap, void * iap) {
  
  sb_argsp sbap = (sb_argsp) ap;
  u1b packet[max_packet_size];
  dl_msgp dmp = (dl_msgp) packet;
  msgp mp = (msgp) packet;
  u1b pkts;
  int j;
  u1b * bp;

  <send the next packet burst>
  }

Defines send_pkts (links are to index).

Fill in some data-message boilerplate.

<send the next packet burst>= (<-U) [D->]

msg_type(mp) = data_msg_type;
dmp->packet_number =
  sbap->nbid->packets_per_buffer - sbap->packets_remaining + 1; 
dmp->buffernumber = sbap->bno;
dmp->max_sequence = sbap->nbid->next_seq_no - 1;
dmp->reserved = sbap->reserved;
bp = sbap->buffer + sbap->nbid->payload_size*(dmp->packet_number - 1);

Figure out how many packets to send in this burst (because send_pkts() had been scheduled to execute, there must be at least one packet available to send). The last packet in the buffer is treated differently from all the other packets in the buffer; otherwise, send_pkts() schedules itself to send the next burst of packets.

<send the next packet burst>+= (<-U) [<-D->]

assert(sbap->packets_remaining > 0);

pkts = minimum(sbap->nbid->nnp.packets_per_burst, sbap->packets_remaining);
sbap->packets_remaining -= pkts;
if (sbap->packets_remaining == 0) pkts--;
else (void) schedule_timer(sbap->nbid->nnp.burst_interval, send_pkts, sbap);

Send 'em off.

<send the next packet burst>+= (<-U) [<-D->]

for (j = 0; j < pkts; j++) {
  msg_length(mp) = 0;
  put_msg_data(mp, bp, sbap->nbid->payload_size);
  send_msg(sbap->nbid, (msgp) packet);
  bp += sbap->nbid->payload_size;
  dmp->packet_number++;
  }

In addition to sending the last packet in the buffer, schedule handlers for resend and ok sub-messages from the receiver.

<send the next packet burst>+= (<-U) [<-D]

if (sbap->packets_remaining == 0) {
  sbap->resend_mhid =
    schedule_mhandler(set_msgtype(0, control), get_resend, ap);
  sbap->ok_mhid =
    schedule_mhandler(set_msgtype(0, control), get_ok, ap);

  msg_type(mp) = ldata_msg_type;
  msg_length(mp) = 0;

  put_msg_data(mp, bp, sbap->nbid->last_payload_size);
  send_msg(sbap->nbid, (msgp) packet);
  }

resend_pkts() sends the next burst of retransmitted packets; ap points to the retransmission information.

<sender.c>+= [<-D->]

static void resend_pkts(void * ap, void * iap) {
  
  resend_datap rsdp = (resend_datap) ap;
  u1b packet[max_packet_size];
  dl_msgp dmp = (dl_msgp) packet;
  msgp mp = (msgp) packet;
  u1b pkts;
  int j;

  <retransmit lost packets>
  }

Defines resend_pkts (links are to index).

Set pkts to the number of packets to send in this interval, which is the smaller of the number of lost packets left to send and the number of packets sent per interval. If there are still packets left to retransmit, resend_pkts() schedules itself to execute in the next interval.

<retransmit lost packets>= (U->) [D->]

pkts = minimum(sba.nbid->nnp.packets_per_burst, rsdp->missing_pkt_cnt);
rsdp->missing_pkt_cnt -= pkts;
if (rsdp->missing_pkt_cnt)
  (void) schedule_timer(sba.nbid->nnp.burst_interval, resend_pkts, rsdp);

Fill in some boilerplate in the retransmitted messages' header.

<retransmit lost packets>+= (U->) [<-D->]

dmp->max_sequence = sba.nbid->next_seq_no - 1;
dmp->buffernumber = sba.bno;
dmp->reserved = sba.reserved;

Retransmit the lost packets. The only tricky bit here is making sure the last packet, if retransmitted, contains the proper amount of data (NETBLT apparently places no constraints on packet-number ordering in retransmission requests).

<retransmit lost packets>+= (U->) [<-D->]

for (j = 0; j < pkts; j++) {
  const u2b pno = rsdp->missing_pkts[rsdp->missing_pkt_cnt + j];
  const bool last_pkt = pno == sba.nbid->packets_per_buffer;

  dmp->packet_number = pno;

  msg_length(mp) = 0;
  msg_type(mp) = (last_pkt ? ldata_msg_type : data_msg_type);
  put_msg_data(mp, sba.buffer + sba.nbid->payload_size*(pno - 1),
    last_pkt ? sba.nbid->last_payload_size : sba.nbid->payload_size);

  send_msg(sba.nbid, (msgp) packet);
  }

Clean up if all packets have been retransmitted.

<retransmit lost packets>+= (U->) [<-D]

if (rsdp->missing_pkt_cnt == 0) {
  free((char *) rsdp->missing_pkts);
  free((char *) rsdp);
  }

If the control message pointed to by iap contains a resend sub-message, copy the missing packet numbers from the sub-message and starting retransmission by calling resend_pkts().

<sender.c>+= [<-D->]

static void get_resend(void * ap, void * iap) {

  sb_argsp sbap = (sb_argsp) ap;
  acqr_msgp cmp = (acqr_msgp) (((msg_bundlep) iap)->mp);
  resend_msgp rmp;

  assert(is_msg_type(cmp, control));
  if ((rmp = (resend_msgp) has_controlm(cmp, resend, sbap))) {

    resend_datap rsdp = (resend_datap) malloc(sizeof(resend_data));
    assert(rsdp);
    rsdp->missing_pkts = (u2b *) malloc(sizeof(u2b)*rmp->pkt_cnt);
    assert(rsdp->missing_pkts);

    rsdp->missing_pkt_cnt = rmp->pkt_cnt;
    byte_copy(rsdp->missing_pkts, ((u1b *) rmp) + sizeof(resend_msg),
              rmp->pkt_cnt*2);
    resend_pkts(rsdp, 0);
    }
  }

Defines get_resend (links are to index).

If the control message pointed to by iap contains an ok sub-message, the current buffer has been successfully transfered and the current send is over.

<sender.c>+= [<-D->]

static void get_ok(void * ap, void * iap) {

  sb_argsp sbap = (sb_argsp) ap;
  acqr_msgp cmp = (acqr_msgp) (((msg_bundlep) iap)->mp);

  assert(is_msg_type(cmp, control));
  if (do_ok_msg((msgp) cmp, sbap)) sbap->done = true;
  }

Defines get_ok (links are to index).

If the control message pointed to by iap has a go sub-message, start sending the contents of the current buffer by calling send_pkts(). Once the go message has been discovered, there's no need for get_go() until the next buffer, so it deletes itself.

<sender.c>+= [<-D->]

static void get_go(void * ap, void * iap) {

  sb_argsp sbap = (sb_argsp) ap;
  acqr_msgp cmp = (acqr_msgp) (((msg_bundlep) iap)->mp);

  assert(is_msg_type(cmp, control));
  if (!has_controlm(cmp, go, sbap)) return;

  cancel_mhandler(sbap->go_mhid);

  send_pkts(ap, 0);
  }

Defines get_go (links are to index).

end_it() handles shut-down time-outs from the receive side; tf() is the termination function for the buffer send.

<sender.c>+= [<-D->]

static void end_it(void * ap, void * iap) {
  ((sb_argsp) ap)->done = true;
  }


static bool tf(void) {
  return sba.done;
  }

Defines end_it (links are to index).

Given nbid, the sending end of a NETBLT connection and buffer, a pointer to a buffer of data, send the data in the buffer to the receiving end of the connection.

<sender.c>+= [<-D]

void send_netblt(netblt_id nbid, u1b * buffer) {

  if (nbid->current_bno == nbid->buffers_per_transfer) {
    fprintf(stderr, "trying to send more data than was negotiated.\n");
    exit(1);
    }

  <send a buffer of data>
  }

Defines send_netblt (links are to index).

Set the buffer number and, if this is the last buffer of the transmission, adjust the transfer parameters.

<send a buffer of data>= (<-U) [D->]

sba.last_buffer = (nbid->current_bno + 1) == nbid->buffers_per_transfer;

if (sba.last_buffer) {
  nbid->nnp.transfer_size =
    nbid->nnp.transfer_size -
    (nbid->buffers_per_transfer - 1)*nbid->nnp.buffer_size;
  nbid->packets_per_buffer =
    ceiling(nbid->nnp.transfer_size, nbid->payload_size);
  nbid->last_payload_size =
    nbid->nnp.transfer_size -
    (nbid->packets_per_buffer - 1)*(nbid->payload_size);
  }

Set up the sending buffer information in sba.

<send a buffer of data>+= (<-U) [<-D->]

sba.nbid = nbid;
sba.buffer = buffer;
sba.bno = ++nbid->current_bno;
sba.packets_remaining = nbid->packets_per_buffer;
sba.nbid->next_seq_no = nbid->next_seq_no;
sba.reserved = (sba.last_buffer ? 1 : 0);
sba.done = false;

Schedule a handler for the go sub-message from the receiving side, then wait for everything to die down.

<send a buffer of data>+= (<-U) [<-D->]

sba.resend_mhid = sba.ok_mhid = 0;
sba.go_mhid = schedule_mhandler(set_msgtype(0, control), get_go, &sba);

run_scheduler(tf);

All packets in the current buffer have been sent and received. Cancel any outstanding message handlers, then wait for the done message from the other side, scheduling a time out in case the receiving end dies.

<send a buffer of data>+= (<-U) [<-D]

if (sba.resend_mhid) cancel_mhandler(sba.resend_mhid);
if (sba.ok_mhid) cancel_mhandler(sba.ok_mhid);

if (sba.last_buffer) {
  timer_id tid;

  sba.done = false;
  tid = schedule_timer(sba.control_ack_timeout*4*1000, end_it, &sba);

  run_scheduler(tf);

  cancel_mhandler(sba.done_mhid);
  cancel_timer(tid);
  }

receive_NETBLT() accepts a pointer to an empty buffer and returns when the buffer's been filled.

<receiver.h>=

#ifndef _receiver_h_defined_
#define _receiver_h_defined_

#include "netblt.h"

extern void receive_netblt(netblt_id nbid, u1b *);

#endif

Defines "netblt.h" (links are to index).

<receiver.c>=

#include "m-handler.h"
#include "timer.h"
#include "bit-array.h"
#include "scheduler.h"
#include "control-sender.h"
#include "msg.h"

<buffer assembly information>

/* Sundries. */

   static ab_args aba;

#  define control_ack_timeout(_a) \
     ceiling((_a)->nbid->packets_per_buffer, (_a)->nbid->nnp.packets_per_burst)*(_a)->nbid->nnp.burst_interval*1000*100

<receiver.c code>

Defines control_ack_timeout (links are to index).

The buffer-assembly routines need to know the following information to do their job: list_start()

<buffer assembly information>= (<-U)

typedef struct {
 u1b         * buffer;
 buffer_number bno;
 netblt_id     nbid;
 bit_array     pkts_rcvd;
 timer_id      rexmit_tid;
 mhandler_id   rcv_mhid, rercv_mhid;
 bool          done;
} ab_args, * ab_argsp;

<receiver.c code>= (<-U) [D->]

static bool buffer_full(ab_argsp abap) {

  if (onescnt_ba(abap->pkts_rcvd) == abap->nbid->packets_per_buffer) {
    cancel_timer(abap->rexmit_tid);
    cancel_mhandler(abap->rcv_mhid);
    cancel_mhandler(abap->rercv_mhid);
    abap->done = true;
    return true;
    }

  return false;
  }

<receiver.c code>+= (<-U) [<-D->]

static void handle_losses(void * ap, void * iap) {

  ab_argsp abap = (ab_argsp) ap;
  resend_msgp rmp;
  u2b * pnop, i;

  if (buffer_full(abap)) return;

  rmp = (resend_msgp) make_msg();
  submsg_type(rmp) = resend_submessage_type;
  sm_buffer_number(rmp) = abap->bno;
  rmp->pkt_cnt = 0;
  pnop = resend_pkt_start(rmp);
  for (i = 1; i <= abap->nbid->packets_per_buffer; i++)
    if (!test_ba(abap->pkts_rcvd, i)) {
      *pnop++ = i;
      rmp->pkt_cnt++;
      }
  send_ctlsdr(rmp, 0);

  abap->rexmit_tid =
    schedule_timer(control_ack_timeout(abap)*1000, handle_losses, abap);
  }

<receiver.c code>+= (<-U) [<-D->]

static void copy_data(ab_argsp abap, void * iap, u2b copy_size) {

  dl_msgp mp = (dl_msgp) (((msg_bundlep) iap)->mp);

  debugp((stderr, "Got a %6s packet, bno = %2d, pno = %2d.\n", msg_name(mp),
          mp->buffernumber, mp->packet_number));

  if (mp->buffernumber != abap->bno) return;
  if ((mp->packet_number < 1) ||
      (mp->packet_number > abap->nbid->packets_per_buffer)) {
    fprintf(stderr, "bad packet number %d received.\n", mp->packet_number);
    return;
    }
  if (test_ba(abap->pkts_rcvd, mp->packet_number)) return;

  byte_copy(abap->buffer + (mp->packet_number - 1)*abap->nbid->payload_size,
            msg_data(mp), copy_size);

  set_ba(abap->pkts_rcvd, mp->packet_number);
  }

<receiver.c code>+= (<-U) [<-D->]

static void receive_sends(void *ap, void * iap) {

  const bool last_pkt = is_msg_type(((msg_bundlep) iap)->mp, ldata);
  ab_argsp abap = (ab_argsp) ap;

  copy_data(abap, iap, last_pkt ? abap->nbid->last_payload_size
                                : abap->nbid->payload_size);
  if (last_pkt) handle_losses(ap, 0);
  }

<receiver.c code>+= (<-U) [<-D->]

static void receive_resends(void *ap, void * iap) {

  ab_argsp abap = (ab_argsp) ap;

  copy_data(abap, iap, is_msg_type(((msg_bundlep) iap)->mp, data)
                                   ? abap->nbid->payload_size
                                   : abap->nbid->last_payload_size);
  (void) buffer_full(abap);
  }

<receiver.c code>+= (<-U) [<-D->]

static bool tf(void) {
  return aba.done;
  }

<receiver.c code>+= (<-U) [<-D]

void receive_netblt(netblt_id nbid, u1b * buffer) {

  submsgp smp;

  if (nbid->current_bno == nbid->buffers_per_transfer) {
    fprintf(stderr, "trying to send more data than was negotiated.\n");
    exit(1);
    }
  <receive_netblt() code>
  }

<receive_netblt() code>= (<-U) [D->]

if (nbid->current_bno == nbid->buffers_per_transfer - 1) {
  nbid->nnp.transfer_size =
    nbid->nnp.transfer_size -
    (nbid->buffers_per_transfer - 1)*nbid->nnp.buffer_size;
  nbid->packets_per_buffer =
    ceiling(nbid->nnp.transfer_size, nbid->payload_size);
  nbid->last_payload_size =
    nbid->nnp.transfer_size -
    (nbid->packets_per_buffer - 1)*(nbid->payload_size);
  }

<receive_netblt() code>+= (<-U) [<-D->]

aba.buffer = buffer;
aba.nbid = nbid;
aba.bno = ++nbid->current_bno;
aba.pkts_rcvd = init_ba(nbid->packets_per_buffer);
aba.done = false;

<receive_netblt() code>+= (<-U) [<-D->]

aba.rcv_mhid =
  schedule_mhandler(set_msgtype(0, data), receive_sends, &aba);
aba.rercv_mhid =
  schedule_mhandler(set_msgtype(0, ldata), receive_resends, &aba);
aba.rexmit_tid =
  schedule_timer(control_ack_timeout(&aba)*1000, handle_losses, &aba);

<receive_netblt() code>+= (<-U) [<-D->]

smp = (submsgp) make_msg();
submsg_type(smp) = go_submessage_type;
sm_buffer_number(smp) = aba.bno;
send_ctlsdr(smp, 0);

<receive_netblt() code>+= (<-U) [<-D->]

run_scheduler(tf);

smp = (submsgp) make_msg();
submsg_type(smp) = ok_submessage_type;
sm_buffer_number(smp) = aba.bno;
((ok_msgp) smp)->burst_size = nbid->nnp.packets_per_burst;
((ok_msgp) smp)->burst_rate = nbid->nnp.burst_interval;
((ok_msgp) smp)->control_timer = control_ack_timeout(&aba);
send_ctlsdr(smp, 0);

<receive_netblt() code>+= (<-U) [<-D]

if (nbid->current_bno == nbid->buffers_per_transfer) {
  msg m;
  run_scheduler(done_ctlsdr);
  msg_type(&m) = done_msg_type;
  msg_length(&m) = 0;
  send_msg(nbid, &m);
  }

free_ba(aba.pkts_rcvd);

<interval.h>=

#ifndef _interval_h_defined_
#define _interval_h_defined_

/* The start of an interval. */

   typedef struct {
     void (* fun)(void *, void *);
     void * arg;
     void * iarg;
     } interval, * interval_ptr;


#define put_interval(_iq, _f, _a, _ia) \
  do {interval_ptr _ip = (interval_ptr) put_clist(_iq); \
      _ip->fun = _f; _ip->arg = _a; _ip->iarg = _ia; } while (false)

#endif

<scheduler.h>=

#ifndef _scheduler_h_defined_
#define _scheduler_h_defined_

#include "defs.h"

extern void
  init_scheduler(int),
  run_scheduler(bool (* f)(void)),
  term_scheduler(void);

#endif

<scheduler.c>= [D->]

#include "scheduler.h"
#include "m-handler.h"
#include "timer.h"
#include "interval.h"


/* Sundries. */

   static c_list old_queue, new_queue, current_queue;

<scheduler.c>+= [<-D->]

void init_scheduler(int skt) {

  old_queue = init_clist(sizeof(interval));
  current_queue = new_queue = init_clist(sizeof(interval));

  init_timer(&current_queue);
  init_mhandler(skt, &current_queue);
  }


void term_scheduler(void) {
  }

<scheduler.c>+= [<-D]

void run_scheduler(bool (* done)(void)) {

  extern int pause(void);

  loop {
    <iterate through the pending events>
    if ((* done)()) break;
    pause();
    }
  }

<iterate through the pending events>= (<-U)

while (!empty_clist(current_queue)) {
  old_queue = current_queue;
  current_queue = new_queue;
  do {
    interval_ptr ip = (interval_ptr) get_clist(old_queue);
    (* ip->fun)(ip->arg, ip->iarg);
    }
  while (!empty_clist(old_queue));
  new_queue = old_queue;
  }

<m-handler interface>= (U->)

extern void
  init_mhandler(int, c_list *),
  term_mhandler(void),
  cancel_mhandler(mhandler_id);

extern mhandler_id
  schedule_mhandler(u2b msg_list, void (* f)(void *, void *), void * arg);

<m-handler.h>=
#include "c-list.h"
#include "msg.h"

#ifndef _m_handler_h_defined_
#define _m_handler_h_defined_

typedef u4b mhandler_id;

#define set_msgtype(_n, _t)    ((_n) | (1 << _t ## _msg_type))
#define is_set_msgtype(_n, _t) ((_n) & (1 << _t ## _msg_type))

typedef struct {
  u4b  host;
  u2b  port;
  msgp mp;
  } msg_bundle, * msg_bundlep;

<m-handler interface>

#endif

<m-handler code>= (U->) [D->]

static msg_bundlep new_mbundle(void) {

  msg_bundlep mbp;

  if (mb_free_list) {
    mbp = (msg_bundlep) mb_free_list;
    mb_free_list = (msg_bundlep) mbp->host;
    }
  else {
    mbp = (msg_bundlep) malloc(sizeof(msg_bundle));
    assert(mbp);
    mbp->mp = make_msg();
    }
  reset_msg(mbp->mp);

  return mbp;
  }

<m-handler code>+= (U->) [<-D->]

static void free_mbundle(msg_bundlep mbp) {
  mbp->host = (u4b) mb_free_list;
  mb_free_list = mbp;
  }


static void fm(void * arg, void * iarg) {
  free_mbundle((msg_bundlep) iarg);
  }

<m-handler code>+= (U->) [<-D->]

static msg_bundlep read_pkt(int skt) {

  msg_bundlep mbp = new_mbundle();
  int i = sizeof(struct sockaddr_in);
  struct sockaddr_in sa;
  int e;

  e = recvfrom(skt, (char *) mbp->mp, max_packet_size, 0,
               (struct sockaddr *) &sa, &i);
  check(e < 0, "recvfrom()");

  mbp->host = ntohl(sa.sin_addr.s_addr);
  mbp->port = ntohs(sa.sin_port);

  debugp((stderr, "Receive %s from %s:%d.\n", message_name(mbp->mp),
          ina2str(mbp->host), mbp->port));

  if (!checksum_ok(mbp->mp, false)) {
    fprintf(stderr, "NETBLT message checksum failed.\n");
    exit(1);
    }
  
  return mbp;
  }

<m-handler code>+= (U->) [<-D->]

static void m_handler(int s) {
  while (input_ready(skt)) {
    msg_bundlep mbp;
    mh_info_ptr mhip;

    mbp = read_pkt(skt);
    
    for (mhip = waiters[msg_type(mbp->mp)]; mhip; mhip = mhip->next)
      put_interval(*event_queuep, mhip->f, mhip->arg, mbp);

    if (waiters[msg_type(mbp->mp)])
      put_interval(*event_queuep, fm, 0, mbp);
    else {
      free_mbundle(mbp);
      debugp((stderr, "Message passed without notice.\n"));
      }
    }
  }

<m-handler code>+= (U->) [<-D->]

void init_mhandler(int s, c_list * eqp) {

  extern void (*sigset (int sig, void (*disp)(int)))(int);

  assert(!event_queuep);
  event_queuep = eqp;

  skt = s;
  mb_free_list = 0;
  free_list = 0;

  set_io_handler(m_handler);
  set_io_int(skt);

  }


void term_mhandler(void) {
  }

<m-handler code>+= (U->) [<-D->]

mhandler_id schedule_mhandler(
  u2b msg_set, void (* f)(void *, void *), void * arg) {

  static u2b id = 0;
  mhandler_id mid = (((u4b) id++) << 16) | (msg_set & 0xffff);
  u1b i;

  assert(msg_set);

  mask_interrupts();

  for (i = 0; i < msg_type_count; i++)
    if (msg_set & (1 << i)) {
      mh_info_ptr mhip;
    
      if (free_list) {
        mhip = free_list;
        free_list = mhip->next;
        }
      else {
        mhip = (mh_info_ptr) malloc(sizeof(mh_info));
        assert(mhip);
        }

      mhip->f = f;
      mhip->arg = arg;
      mhip->mid = mid;
      mhip->next = waiters[i];
      waiters[i] = mhip;
      }

  unmask_interrupts();

  return mid;
  }

<m-handler code>+= (U->) [<-D->]

static mh_info_ptr free_mhi(mh_info_ptr mhip, mhandler_id mid) {

  mh_info h;
  mh_info_ptr this, last;

  last = &h;
  this = last->next = mhip;

  while (this) {
    if (this->mid == mid) {
      last->next = this->next;
      this->next = free_list;
      free_list = this;
      }
    else
      last = this;
    this = last->next;
    }

  return h.next;
  }

<m-handler code>+= (U->) [<-D]

void cancel_mhandler(mhandler_id mid) {

  u2b msg_set = (u2b) mid;
  u1b i;

  mask_interrupts();

  for (i = 0; i < msg_type_count; i++)
    if (msg_set & (1 << i)) waiters[i] = free_mhi(waiters[i], mid);

  unmask_interrupts();
  }

<m-handler.c>=

#include <signal.h>
#include <unistd.h>
#include <sys/time.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include "m-handler.h"
#include "interval.h"
#include "system.h"


/* The message-handler data structure. */

   typedef struct Mh_info {
     microsecs wake_up;
     void          (* f)(void *, void *);
     void           * arg;
     mhandler_id      mid;
     struct Mh_info * next;
     } mh_info, * mh_info_ptr;

   static mh_info_ptr waiters[msg_type_count];


/* Sundires */

   static mh_info_ptr free_list;

   static int skt;

   static c_list * event_queuep;

   static msg_bundlep mb_free_list;


<m-handler code>

<timer.h>=

#ifndef _timer_h_defined_
#define _timer_h_defined_

#include "c-list.h"

typedef struct Timer_id * timer_id;

extern void
  init_timer(c_list * eqp),
  cancel_timer(timer_id),
  term_timer(void);

extern timer_id
  schedule_timer(microsecs, void (* f)(void *, void *), void * arg);

#endif

<timer code>= (U->) [D->]

static void schedule_next(void) {

  if (pending.next) set_timer(pending.next->wake_up);
  }

<timer code>+= (U->) [<-D->]

static microsecs get_timer(void) {

  struct itimerval t;

  if (getitimer(ITIMER_REAL, &t) == -1) {
    perror("getitimer() failed");
    exit(1);
    }

  return (t.it_value.tv_sec*1000000 + t.it_value.tv_usec);
  }

<timer code>+= (U->) [<-D->]

static void timer_handler(int s) {

  timer_id tid;

  mask_interrupts();

  for (tid = pending.next; tid; tid = tid->next)
    tid->wake_up -= pending.next->wake_up;

  while (pending.next && (pending.next->wake_up == 0)) {
    interval_ptr ip = (interval_ptr) put_clist(*event_queuep);
    ip->fun = pending.next->f;
    ip->arg = pending.next->arg;
    ip->iarg = 0;

    tid = pending.next->next;
    pending.next->next = free_list;
    free_list = pending.next;
    pending.next = tid;
    }

  schedule_next();

  unmask_interrupts();
  }

<timer code>+= (U->) [<-D->]

void init_timer(c_list * eqp) {

  extern void (*sigset (int sig, void (*disp)(int)))(int);

  assert(!event_queuep);
  event_queuep = eqp;

  set_timer_handler(timer_handler);

  }


void term_timer(void) {

  }

<timer code>+= (U->) [<-D->]

timer_id schedule_timer(
  microsecs wake_up, void (* f)(void *, void *), void * arg) {

  timer_id tid, next, prev;

  mask_interrupts();

  <insert the new time-out interval>

  <reschedule the timer if necessary>

  unmask_interrupts();

  return tid;
  }

<insert the new time-out interval>= (U->)

if (free_list) {
  tid = free_list;
  free_list = free_list->next;
  }
else {
  tid = (timer_id) malloc(sizeof(Timer_id));
  assert(tid);
  }

tid->wake_up = wake_up;
tid->f = f;
tid->arg = arg;

prev = &pending;
next = prev->next;
while (next && (tid->wake_up >= next->wake_up)) {
  prev = next;
  next = next->next;
  }
tid->next = next;
prev->next = tid;

<reschedule the timer if necessary>= (U->)

if (pending.next == tid) {
  if (tid->next && (tid->wake_up != tid->next->wake_up)) {
    microsecs spent = get_timer();
    timer_id tid2;

    set_timer(0);
    for (tid2 = tid->next; tid2; tid2 = tid2->next) tid2->wake_up -= spent;
    }
  schedule_next();
  }

<timer code>+= (U->) [<-D]

void cancel_timer(timer_id tid) {

  timer_id next, prev;

  mask_interrupts();

  prev = &pending;
  next = prev->next;
  while (next && (tid != next)) {
    prev = next;
    next = next->next;
    }

  if (next) {
    prev->next = tid->next;

    <if necessary, reschedule the timer>

    tid->next = free_list;
    free_list = tid;
    }

  unmask_interrupts();
  }

<if necessary, reschedule the timer>= (U->)

if (prev == &pending) {
  microsecs spent = get_timer();
  set_timer(0);
  if (tid->next && (tid->wake_up != tid->next->wake_up)) {
    timer_id tid2;
    for (tid2 = pending.next; tid2; tid2 = tid2->next)
      tid2->wake_up -= spent;
    schedule_next();
    }
  }

<timer.c>=

#include <signal.h>
#include <unistd.h>
#include <sys/time.h>
#include "timer.h"
#include "interval.h"
#include "system.h"


/* The timer data structure. */

   typedef struct Timer_id {
     microsecs wake_up;
     void      (* f)(void *, void *);
     void    * arg;
     timer_id  next;
     } Timer_id;

   static Timer_id pending;


/* Sundires */

   static timer_id free_list;

   static c_list * event_queuep;

<timer code>

<control-sender.h>=

#ifndef _control_sender_h_defined_
#define _control_sender_h_defined_

#include "netblt.h"

extern void
  init_ctlsdr(netblt_id),
  send_ctlsdr(void *, void *),
  term_ctlsdr(void);

extern bool
  done_ctlsdr(void);

#endif

<control sender code>= (U->) [D->]

static void seqno_handler(void * ap, void * iap) {

  msgp mp = (msgp) (((msg_bundlep) iap)->mp);
  sequence_number sno;

  sno = (is_msg_type(mp, nullack) ? ((n_msgp) mp)->max_sequence
                                  : ((dl_msgp) mp)->max_sequence);
  
  if (max_rcvd_sno < sno) {
    max_rcvd_sno = sno;
    <clear acknowledged control messages>
    }
  }

<clear acknowledged control messages>= (<-U)

do {
  submsgp smp = *((submsgp *) head_clist(pending_ctl_msgs));
  if (smp->sequence_number > sno) break;
  free_msg((msgp) (*((submsgp *) get_clist(pending_ctl_msgs))));
  }
while (!empty_clist(pending_ctl_msgs));

if (empty_clist(pending_ctl_msgs)) {
  cancel_timer(tid);
  tid = 0;
  }

<control sender code>+= (U->) [<-D->]

void init_ctlsdr(netblt_id n) {
  max_sent_sno = max_rcvd_sno = 0;
  nbid = n;
  pending_ctl_msgs = init_clist(sizeof(submsgp));
  timer_interval = 10000000;
  mhid = schedule_mhandler(
           set_msgtype(set_msgtype(set_msgtype(0, data), ldata), nullack),
           seqno_handler, 0);
  }

<control sender code>+= (U->) [<-D->]

void send_ctlsdr(void * ap, void * iap) {

  submsgp smp = (submsgp) ap;
  submsgp * smpp;               /* must zero test before dereference. */
  u1b * bp, * end;
  msgp mp = make_msg();

  if (smp) {
    smp->sequence_number = ++max_sent_sno;
    *((submsgp *) put_clist(pending_ctl_msgs)) = smp;
    if (tid) cancel_timer(tid);
    }

  msg_type(mp) = control_msg_type;
  bp = msg_data(mp);
  end = ((u1b *) mp) + max_packet_size;
  initi_clist(pending_ctl_msgs);

  while ((smpp = (submsgp *) nexti_clist(pending_ctl_msgs))) {
    const u2b l = smsg_length(*smpp);
    if (bp + l > end) break;
    byte_copy(bp, *smpp, l);
    bp += l;
    }

  msg_length(mp) = bp - msg_data(mp) + msg_header_size(mp);
  send_msg(nbid, mp);
  free_msg(mp);

  tid = schedule_timer(timer_interval, send_ctlsdr, 0);
  }

<control sender code>+= (U->) [<-D->]

bool done_ctlsdr(void) {
  return max_rcvd_sno == max_sent_sno;
  }

<control sender code>+= (U->) [<-D]

void term_ctlsdr(void) {
  if (tid) cancel_timer(tid);
  tid = 0;
  }

<control-sender.c>=

#include "control-sender.h"
#include "timer.h"
#include "m-handler.h"
#include "msg.h"

/* Sundries. */

   static timer_id tid;
   static mhandler_id mhid;
   static netblt_id nbid;
   static sequence_number max_rcvd_sno;
   static sequence_number max_sent_sno;
   static c_list pending_ctl_msgs;
   static microsecs timer_interval;

<control sender code>

<system.h>=

#ifndef _set_io_int_h_
#define _set_io_int_h_

#include "defs.h"

extern void set_io_int(int), mask_interrupts(void), unmask_interrupts(void),
  set_io_handler(), set_timer_handler(), set_timer(microsecs), 
  bind_skt(int, u4b, u2b);

extern bool input_ready(int);

extern int get_host_name(char *, u2b);

#endif

<sunos.c>=

#include <sys/types.h>
#include <sys/time.h>
#include <fcntl.h>
#include <stropts.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <signal.h>
#include <errno.h>
#include "defs.h"
#include "system.h"

extern int sigblock(int);


void set_io_int(int fd) {
  check(fcntl(fd, F_SETOWN, getpgrp()), "fcntl(f_setown)");
  check(fcntl(fd, F_SETFL, FASYNC), "ioctl(f_setfl, fasync)");
  }


bool input_ready(int fd) {

  struct timeval t;
  fd_set r;
  int e;
  extern int select(int, fd_set *, fd_set *, fd_set *, struct timeval *);

  if (fd < 0) return false;

  bzero((char *) &t, sizeof(t));
  FD_ZERO(&r);
  FD_SET(fd, &r);

  e = select(fd + 1, &r, NULL, NULL, &t);
  if ((e < 0) && (errno == EBADF)) return 0;
  check(e < 0, "select()");

  return e;
  }


void mask_interrupts() {
  (void) sigblock(sigmask(SIGIO) | sigmask(SIGALRM));
  }


void unmask_interrupts () {
  extern int sigsetmask(int);
  (void) sigsetmask(sigblock(0) & ~(sigmask(SIGIO) | sigmask(SIGALRM)));
  }


void set_io_handler(void (* h)(int)) {
  check(signal(SIGIO, h) == SIG_ERR, "sigset(sigio)");
  }


void set_timer_handler(void (* h)(int)) {
  check(signal(SIGALRM, h) == SIG_ERR, "sigset(sigalrm)");
  }


void set_timer(microsecs i) {
  extern int ualarm(int, int);
  ualarm(i, 0);
  }


int get_host_name(char * hostname, u2b hostname_size) {
  return (gethostname(hostname, (int) hostname_size) ? errno : 0;
  }


void bind_skt(int skt, u4b host, u2b port) {

  struct sockaddr_in sa;
  const int sas = sizeof(struct sockaddr_in);
  extern int bind(int, struct sockaddr *, int);

  bzero((char *) &sa, sas);
  sa.sin_family = AF_INET;
  sa.sin_addr.s_addr = htonl(host);
  sa.sin_port = htons(port);

  check(bind(skt, (struct sockaddr *) &sa, sas), "bind()");
  }

<solaris.c>=
#include <sys/types.h>
#include <fcntl.h>
#include <stropts.h>
#include <sys/socket.h>
#include <sys/ioctl.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <signal.h>
#include <unistd.h>
#include <errno.h>
#include "defs.h"
#include "system.h"

extern int sighold(int), sigrelse(int);
extern void (*sigset (int sig, void (*disp)(int)))(int);

void set_io_int(int fd) {
  check(fcntl(fd, F_SETOWN, getpgrp()), "fcntl(f_setown)");
  check(ioctl(fd, I_SETSIG, S_RDNORM), "ioctl(i_setsig, s_rdnorm)");
  }


bool input_ready(int fd) {
  int e, i;
  check((e = ioctl(fd, I_NREAD, &i)) < 0, "ioctl(I_NREAD)");
  return e;
  }


void mask_interrupts() {
  check(sighold(SIGPOLL), "sighold(POLL)");
  check(sighold(SIGALRM), "sighold(ALRM)");
  }


void unmask_interrupts () {
  check(sigrelse(SIGALRM), "sigrelse(ALRM)");
  check(sigrelse(SIGPOLL), "sigrelse(POLL)");
  }


void set_io_handler(void (* h)(int)) {
  check(sigset(SIGIO, h) == SIG_ERR, "sigset(sigio)");
  }


void set_timer_handler(void (* h)(int)) {
  check(sigset(SIGALRM, h) == SIG_ERR, "sigset(sigalrm)");
  }


void set_timer(microsecs i) {
  ualarm(i, 0);
  }

int get_host_name(char * hostname, u2b hostname_size) {
  return (gethostname(hostname, (int) hostname_size) ? errno : 0);
  }


void bind_skt(int skt, u4b host, u2b port) {

  struct sockaddr_in sa;
  const int sas = sizeof(struct sockaddr_in);

  bzero((char *) &sa, sas);
  sa.sin_family = AF_INET;
  sa.sin_addr.s_addr = htonl(host);
  sa.sin_port = htons(port);

  check(bind(skt, (struct sockaddr *) &sa, sas), "bind()");
  }

<c-list.h>=

#ifndef _c_list_h_defined_
#define _c_list_h_defined_

#include "defs.h"

typedef struct C_list * c_list;

extern c_list init_clist(u2b);
extern void
   term_clist(c_list),
 * head_clist(c_list),
 * put_clist(c_list),
 * get_clist(c_list),
   initi_clist(c_list),
 * nexti_clist(c_list);

extern bool empty_clist(c_list);

#endif

<c-list data structures>= (U->) [D->]

typedef struct C_list {
  u1b * head, * tail;
  u2b ring_size, size, item_size;
  u1b * cursor;
  u2b icnt;
  } C_list;

<c-list data structures>+= (U->) [<-D]

#  define next_b(_b) ((u1b **) (_b))[0]
#  define prev_b(_b) ((u1b **) (_b))[1]
#  define data_o(_b) ((_b) + 8)

<c-list code>= (U->) [D->]

static void grow(c_list cl) {

  u1b * i = (u1b *) malloc(cl->item_size + 8);
  assert(i);

  prev_b(i) = prev_b(cl->tail);
  next_b(prev_b(cl->tail)) = i;

  next_b(i) = cl->tail;
  prev_b(cl->tail) = i;

  cl->tail = i;
  cl->ring_size++;
  }

<c-list code>+= (U->) [<-D->]

c_list init_clist(u2b item_size) {

  c_list cl;

  assert(item_size > 0);

  cl = (c_list) malloc(sizeof(C_list));
  assert(cl);
  cl->head = cl->tail = (u1b *) malloc(item_size + 8);
  prev_b(cl->head) = cl->head;
  next_b(cl->head) = cl->head;

  cl->ring_size = 1;
  cl->item_size = item_size;
  cl->size = 0;

  return cl;
  }

<c-list code>+= (U->) [<-D->]

void term_clist(c_list cl) {
  free((char *) cl);
  }

<c-list code>+= (U->) [<-D->]

void * put_clist(c_list cl) {

  u1b * i;

  if (cl->size == cl->ring_size) grow(cl);
  i = data_o(cl->tail);
  cl->tail = next_b(cl->tail);
  cl->size++;
  
  return (void *) i;
  }

<c-list code>+= (U->) [<-D->]

void * get_clist(c_list cl) {

  u1b * i;

  if (cl->size == 0) {
    fprintf(stderr, "Doing a get on an empty c-list.\n");
    exit(1);
    }

  i = data_o(cl->head);
  cl->head = next_b(cl->head);
  cl->size--;

  return i;
  }

<c-list code>+= (U->) [<-D->]

void * head_clist(c_list cl) {

  if (cl->size == 0) {
    fprintf(stderr, "Getting the head of an empty c-list.\n");
    exit(1);
    }

  return data_o(cl->head);
  }

<c-list code>+= (U->) [<-D->]

bool empty_clist(c_list cl) {
  return (cl->size == 0);
  }

<c-list code>+= (U->) [<-D->]

void initi_clist(c_list cl) {
  cl->cursor = cl->head;
  cl->icnt = 0;
  }

<c-list code>+= (U->) [<-D]

void * nexti_clist(c_list cl) {

  if (cl->icnt == cl->size) return 0;
  cl->cursor = next_b(cl->cursor);
  cl->icnt++;

  return data_o(prev_b(cl->cursor));
  }

<c-list.c>=

#include "c-list.h"

<c-list data structures>

<c-list code>

<bit array external definitions>= (U->)

typedef struct Bit_array * bit_array;

extern bit_array init_ba(u2b);

extern void 
  free_ba(bit_array),
  set_ba(bit_array, u2b),
  clear_ba(bit_array, u2b),
  zeros_ba(bit_array, u2b *, u2b);

extern bool test_ba(bit_array, u2b);

extern u2b
  onescnt_ba(bit_array ba),
  size_ba(bit_array);

<bit-array.c>= [D->]

#include "bit-array.h"

struct Bit_array {
  u2b   byte_count;
  u2b   bit_count;
  u1b * bits;
  u2b   ones_count;
  };

<bit-array.c>+= [<-D->]

bit_array init_ba(u2b size) {

  bit_array ba = (bit_array) malloc(sizeof(struct Bit_array));
  assert(ba);
  assert(size > 0);
  ba->bit_count = size;
  ba->byte_count = ceiling(size, 8);
  ba->ones_count = 0;
  ba->bits = (u1b *) malloc(sizeof(u1b)*ba->byte_count);
  assert(ba->bits);
  bzero((char *) (ba->bits), ba->byte_count);
  ba->bits[ba->byte_count - 1] = 0xff << (8 - (ba->byte_count*8 - size));

  return ba;
  }

<bit-array.c>+= [<-D->]

void free_ba(bit_array ba) {
  free((char *) ba->bits);
  free((char *) ba);
  }

<bit-array.c>+= [<-D->]

u2b size_ba(bit_array ba) {
  return ba->bit_count;
  }

<bit-array.c>+= [<-D->]

#define check_bno(_ba, _bno, _w) \
  do if ((0 == (_bno)) || (_ba->bit_count < (_bno))) { \
    fprintf(stderr,"bit number %d in %s_ba() not in [1..%d]\n",\
            (_bno), #_w, _ba->bit_count); \
    exit(1); } while (false)

<bit-array.c>+= [<-D->]

void set_ba(bit_array ba, u2b bno) {

  check_bno(ba, bno, set);
  if (!test_ba(ba, bno)) ba->ones_count++;
  ba->bits[(bno - 1) >> 3] |= (1 << ((bno - 1) & 7));
  }

<bit-array.c>+= [<-D->]

void clear_ba(bit_array ba, u2b bno) {

  check_bno(ba, bno, clear);
  if (test_ba(ba, bno)) ba->ones_count--;
  ba->bits[(bno - 1) >> 3] &= ~(1 << ((bno - 1) & 7));
  }

<bit-array.c>+= [<-D->]

bool test_ba(bit_array ba, u2b bno) {

  check_bno(ba, bno, test);
  return (ba->bits[(bno - 1) >> 3] & (1 << ((bno - 1) & 7)));
  }

<bit-array.c>+= [<-D->]

u2b onescnt_ba(bit_array ba) {
  return ba->ones_count;
  }

<bit-array.c>+= [<-D->]

static void put_zeros(u1b bits, u2b ** datap, u2b * sizep, u2b offset) {
  
  u1b i;

  for (i = 0; (i < 8) && (*sizep > 0); i++) {
    if (!(bits & 1)) {
      *(*datap++) = offset + i;
      (void) (*sizep--);
      }
    bits >>= 1;
    }
  }

<bit-array.c>+= [<-D]

void zeros_ba(bit_array ba, u2b * data, u2b size) {

  u1b offset = 1;
  u1b i;

  for (i = 0; (i < ba->byte_count) && (size > 0); i++) {
    if (ba->bits[i] != 0xff) put_zeros(ba->bits[i], &data, &size, offset);
    offset += 8;
    }
  }

<bit-array.h>=

#ifndef _bit_array_h_defined_
#define _bit_array_h_defined_

#include "defs.h"

<bit array external definitions>

#endif

<stop-watch external definitions>= (U->)

typedef struct stop_watch * stop_watch;

extern stop_watch init_sw(void);
extern void term_sw(stop_watch);
extern void start_sw(stop_watch);
extern microsecs stop_sw(stop_watch);

<stop-watch.c>= [D->]

#include "stop-watch.h"
#include <sys/time.h>

struct stop_watch {
  struct timeval start;
  };

#define get_time(_t) \
  if (gettimeofday(&(_t), NULL)) perror("gettimeofday() failed")

<stop-watch.c>+= [<-D->]

stop_watch init_sw(void) {
  stop_watch sw = (stop_watch) malloc(sizeof(struct stop_watch));
  assert(sw);
  return sw;
  }

<stop-watch.c>+= [<-D->]

void term_sw(stop_watch sw) {
  bzero((char *) sw, sizeof(struct stop_watch));
  free((char *) sw);
  }

<stop-watch.c>+= [<-D->]

void start_sw(stop_watch sw) {
  get_time(sw->start);
  }

<stop-watch.c>+= [<-D]

microsecs stop_sw(stop_watch sw) {
  struct timeval stp;
  get_time(stp);
  return ((stp.tv_sec - sw->start.tv_sec)*1000000 + stp.tv_usec) - sw->start.tv_usec;
  }

<stop-watch.h>=
#ifndef _stop_watch_h_defined_
#define _stop_watch_h_defined_

#include "defs.h"

<stop-watch external definitions>

#endif

<keep-list.h>=

#ifndef _keep_list_h_defined_
#define _keep_list_h_defined_

#include "defs.h"

typedef struct Keep_list * keep_list;

extern keep_list init_kl(void);
extern void free_kl(keep_list);
extern void add_kl(keep_list, void *);
extern void drop_kl(keep_list, void *);

#endif

<keep-list.c>= [D->]

#include "keep-list.h"

struct Keep_list {
  u2b next, max;
  void ** list;
  };

<keep-list.c>+= [<-D->]

keep_list init_kl(void) {

  keep_list kl = (keep_list) malloc(sizeof(struct Keep_list));

  kl->next = 0;
  kl->max = 16;
  kl->list = (void **) malloc(sizeof(void *)*kl->max);
  assert(kl->list);

  return kl;
  }

<keep-list.c>+= [<-D->]

void free_kl(keep_list kl) {
  free((char *) kl->list);
  free((char *) kl);
  }

<keep-list.c>+= [<-D->]

void add_kl(keep_list kl, void * e) {

  if (kl->next >= kl->max) {
    void ** nl = (void **) malloc(sizeof(void *)*kl->max*2);
    assert(nl);
    byte_copy((char *) nl, (const char *) kl->list, sizeof(void *)*kl->max);
    kl->max *=2;
    free((char *) kl->list);
    kl->list = nl;
    }

  kl->list[kl->next++] = e;
  }

<keep-list.c>+= [<-D]

void drop_kl(keep_list kl, void * e) {

  u1b i;

  for (i = 0; i < kl->next; i++) 
    if (kl->list[i] == e) {
      kl->list[i] = kl->list[--kl->next];
      return;
      }

  fprintf(stderr, "keep-list drop failure.\n");
  exit(1);
  }

<defs.h>=

#ifndef _defs_h_defined_
#define _defs_h_defined_

#include <assert.h>
#include <stdio.h>
#include <unistd.h>

typedef unsigned char u1b;
typedef unsigned short u2b;
typedef unsigned long u4b;

typedef u4b buffer_number;
typedef u2b sequence_number;

typedef u2b millisecs;
typedef u2b seconds;
typedef u4b microsecs;

typedef u1b bool;
#define true 1
#define false 0

#define loop while (true)
#define byte_copy(_d, _s, _c) memcpy((char *) (_d), (const char *) (_s), (_c))
#define ceiling(_a, _b) ((_a)/(_b) + (((_a)/(_b))*(_b) == (_a) ? 0 : 1))
#define minimum(_a, _b) ((_a) < (_b) ? (_a) : (_b))

#define check(_t, _w) \
   do if (_t) {perror(_w " failure"); { before_exit() ; exit(1); }} while (0)

#define set_sig(_s, _sh) \
   check((sigset(_s, _sh) == SIG_ERR), "sigset(" #_s ")")

#define str2ina(_s) ntohl(inet_addr(_s))

#ifdef Debugp
# define debugp(_a) fprintf _a
#else
# define debugp(_a)
#endif
  

extern char * malloc(int);
extern void bzero(char *, int);
extern void free(char *), before_exit(void);
extern u2b port(int);
extern u4b skt_addr(int, char *), host_addr(char *);
extern char * ina2str(u4b);
extern int fprintf(FILE *, const char *, ...);
extern int printf(const char *, ...);
extern void perror(const char *);
extern int sscanf(const char *, const char *, ...);
extern int close(int);

typedef struct Netblt_id * netblt_id;

#endif

<utility code>= (U->) [D->]

char * ina2str(u4b ina) {
  struct in_addr ia;
  ia.s_addr = htonl(ina);
  return inet_ntoa(ia);
  }

<utility code>+= (U->) [<-D->]

u4b skt_addr(int skt, char * buff) {
  int sa = sizeof(struct sockaddr_in);
  struct sockaddr_in a;
  if (getsockname(skt, (struct sockaddr *) &a, &sa)) return 0;
  if (buff) strcpy(buff, ina2str(a.sin_addr.s_addr));
  return  a.sin_addr.s_addr;
  }

<utility code>+= (U->) [<-D->]

u4b host_addr(char * buff) {

  u4b hst;
  char name[100];
  struct hostent * h;

  if (get_host_name(name, sizeof(name))) return 0;

  h = gethostbyname(name);
  if (!h) return 0;

  hst = ntohl(*((u4b *) (h->h_addr)));
  if (buff) strcpy(buff, ina2str(hst));

  return hst;
  }

<utility code>+= (U->) [<-D]

u2b port(int skt) {
  int sa = sizeof(struct sockaddr_in);
  struct sockaddr_in a;
  if (getsockname(skt, (struct sockaddr *) &a, &sa)) return 0;
  return ntohs(a.sin_port);
}

<utils.c>=

#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <netdb.h>
#include "defs.h"
#include "system.h"

<utility code>

void before_exit(void) {
  }