nettee(1)

📦 version 0.3.5, in nixos 26.05, nixpkgs unstable

✨ nettee(1) ✨ — fresh from the nettee Manual 💅

Okay bestie, this is the nettee(1) page from the nettee Manual, and we're starting with the interface for nettee_cmd — the little protocol child process vibe. No cap, every detail is staying. 🫶

🌸 NAME 🌸

nettee_cmd — the interface between nettee and programs run by it with the -process command line option.

🦋 DESCRIPTION 🦋

The nettee -process option starts the specified command as a child process and then communicates with it through the simple protocol described here. In addition, the command may contain command line options. When -process is not specified nettee simply copies from the input to the output(s) and does not create or communicate with a child process. If -process is specified the command may modify the data stream and direct some parts to -next and others to -output. For instance, it can peel off a subset of data to the local machine and echo the rest down the chain. Additionally the command could create or destroy data.

The program started by -process communicates with nettee solely through stdin and stdout. Commands are read from stdin and status messages are written to stdout. Both of these are newline terminated strings. Status messages all begin with a number, shown below as X16X, which indicates a 16 character hex formatted long long unsigned integer value. Cases must be as shown.

📥 Commands read by the child from stdin 💌

  • STATUS S — Emit the current status at least every S seconds.
    • These status messages are in ADDITION to any messages which might be sent due to events or other commands. These messages allow the parent to determine if the child has failed.
    • May appear once before START. Otherwise, no additional status messages are sent.
    • Return status READY or RUNNING.
  • STREAMS N — Number of streams which will be created.
    • Must be sent once before START.
    • Return status READY.
  • IOP SIZE MINFREE NET_NW NET_XR NET_XW LCL_NW LCL_XR LCL_XW — Set the IO parameters and buffer sizes (all values are in bytes).
    • SIZE — Set the IO buffer size.
    • MINFREE — Sets the minimum contiguous free space in an input ring buffer which must exist before a read will be attempted. Default 1.
    • NET_NW — (NETwork_miNWrite) sets the minimum available data before a network write will be attempted. Default 1.
    • NET_XR — (NETwork_maXRead) sets the maximum size of a single network read operation. Default 131072.
    • NET_XW — (NETwork_maXWrite) sets the maximum size of a single network write operation. Default 131072.
    • LCL_NW — (lcl_miNWrite) sets the minimum available data before a local write will be attempted. Default 1.
    • LCL_XR — (lcl_maXRead) sets the maximum size of a single local read operation. Default 131072.
    • LCL_XW — (lcl_maXWrite) sets the maximum size of a single local write operation. Default 131072.
    • Min values must be >=1 and <= buffer size.
    • Max values must be >=1.
    • May appear once before START. Otherwise, the defaults are used.
    • Return status READY.
  • IO I name mask — Define input stream stream I from unix fd (if name is an integer) or named socket or file "name" otherwise.
    • Mask bits are (one of: 1=In, 2=Out) PLUS (one of: 4=Local, 8=Network).
    • Must appear once per IO stream, after STREAMS and before START.
    • Return status READY.
  • START — Begin or resume reading data.
    • Must appear after the IO Streams are configured..
    • Return status RUNNING.
  • PAUSE — Stop reading and writing data.
    • Continue to monitor stdin for further commands.
    • May appear after START.
    • Return status READY.
  • DISABLE I — Stop reading or writing data on the indicated stream and continue running.
    • Along with ERROR messages sent up by the process allows nettee to implement Continue On [Network/Local][Read/Write] Failure.
    • (Continue on Read Failure MAY makes sense with -process [but probably not].)
    • May appear after START.
    • Return status READY or RUNNING.
  • EXIT — On receiving this signal the client must exit() immediately.
    • May appear once at any time.
    • Return status (NONE).

📤 Status messaages written by the child to stdout 📤

  • X16X READY — Ready to run at X16X bytes READ from all input streams.
    • The count does not include bytes from the command stream.
    • Also emitted whenever a command is processed without error.
  • X16X RUNNING — Running at X16X bytes READ from all input streams.
    • Response to a START, also may be a periodic status message.
  • X16X FAILED type message — A fatal event has been detected. The child process exits after sending the message.
    • type values are:
      • COMMAND: the command was not understood or could not be carried out, for instance, an open() failed.
      • DATA: some problem was found in the input data.
      • OTHER: any other unrecoverable error, for instance, a failure to allocate more memory.
    • message may contain one line of additional information.
  • X16X DONE type stream — Something completed. If it is an IO stream then that stream has closed.
    • Type + stream are:
      • COMMAND 0: X16X is the number of commands processed.(from stream 0).
      • IO I: X16X is the number of bytes read or written on stream I.
      • ALL 0: X16X is the total number of bytes written.
  • X16X ERROR type stream — A (possibly) recoverable IO error has taken place.
    • Types are READ or WRITE. The X16X is the number of bytes read/written on that stream.

🌀 RB functions 🌀

Okay so the files rb.c and rb.h are where all the ring buffer functions live, plus the definitions for the RB symbols that get used in the examples all over this man page and elsewhere in nettee. ✨

The whole ring buffer organization is essentially a variation of Simon Cooke's "Bip Buffer" method, which was published right here:

http://www.codeproject.com/KB/IP/bipbuffer.aspx?df=100&forumid=14010&exp=0&select=494356

with the title "The Bip Buffer - The Circular Buffer with a Twist". The whole point, bestie, is making most reads into the ring buffer continuous — no awkward splitting. Some of the functions take a minfree parameter: if at the end of an operation that moves data into the ring buffer the top of data ends up less than minfree away from the top of the data buffer, a gap gets introduced. That way the next data transfer into the buffer can start at the bottom of the data buffer and won't need to be split. 🫶

🚨 Changing the value of minfree while a ring buffer is in use is not supported. Don't do it, it's not that kind of party.

The RB_NEXUS data structure is literally this:

typedef struct ring_buffer     RB_NEXUS;
struct ring_buffer {
  char *buf;                /* ring buffer data area                                 */
  int   blen;               /* amount of data held within  data buffer)              */
  int   data;               /* the start of the data in a ring buffer                */
  int   size;               /* maximum storage capacity of the buffer                */
  int   gap;                /* if not zero, indicates a gap <MINFREE at top of buffer*/
};

The RB functions are, in order, and I'm gonna walk you through every single one:

🎀 int rb_adjust(RB_NEXUS *rb, int count);

Adjust the ring buffer as if count bytes had been removed from the trailing edge of the data region. Returns 0 for valid, or -1 if count exceeds the number of bytes actually present.

🧚 void rb_clear(RB_NEXUS *rb, int fill);

Fill a ring buffer with the supplied value (typically "0"). Also resets the data, gap, and blen values to zero. Fresh start energy. 💫

🌸 RB_NEXUS *rb_create(int size, int count);

Create one or more empty ring buffers and get a pointer to them back. If it fails, it cleans up the ones it already created (if any) and returns NULL. Responsible king behavior, ngl.

🦋 void rb_delete(RB_NEXUS **rb, int count);

Delete one or more empty ring buffers that were allocated in a block with rb_create. Releases all the memory and returns a NULL pointer. The value of count must be the same for rb_delete as it was for the corresponding rb_create — like, all buffers created by a single rb_create have to be removed with a single rb_delete. Matching set, no mixing. 💅

💜 int rb_flat_memcmp(RB_NEXUS *s1, char *s2, int n);

Compare up to n bytes of data in one ring buffer and one flat buffer. Returns:

  • <0 if s1 < s2,
  • >0 if s1 > s2,
  • 0 if s1 = s2.

If either buffer pointer is invalid, a fatal error occurs and the program exits. RIP. 💀

Let size refer to the amount of data in s1 and the string length of s2. Here's the fine print:

  • If s1 and s2 are different sizes, and are identical up to the size of s1, and s1 is less than n characters, then -1 gets returned.
  • If s1 and s2 are the same size, and n is larger than that, they'll only be compared up to the amount of data they contain.
  • If s2 is shorter than s1, and s2 is less than n characters, the result of the function is undefined. (The most likely outcome is a segfault.) 😭 So maybe don't.
Note that the comparison is done as unsigned char, even though the rb structure holds data as char.

🍓 int rb_rb_memcmp(RB_NEXUS *s1, RB_NEXUS *s2, int n);

Compare up to n bytes of data in two ring buffers. Returns:

  • <0 if s1 < s2,
  • >0 if s1 > s2,
  • 0 if s1 = s2.

If either ring buffer pointer is invalid, a fatal error occurs and the program exits. Same deal as above, bestie.

Let size refer to the amount of data in s1 or s2:

  • If the size of s1 is larger than the size of s2, and they're identical up to the size of s2, and the size of s2 is less than n characters, then +1 gets returned.
  • If the size of s1 is less than the size of s2, and they're identical up to the size of s1, and the size of s1 is less than n characters, then -1 gets returned.
  • If the sizes of s1 and s2 are the same, and n is larger than that, they'll only be compared up to the amount of data they contain.
Again: comparison is done as unsigned char, even though the rb structure holds data as char. Not a typo, that's just how it is. 🎀

🌈 int rb_from_flat(RB_NEXUS *dst, char *array, int n, int minfree);

Copy up to n bytes of data from a flat array into a ring buffer, in 1 or 2 memcpy calls. Efficient icon. ✨

Returns the number of bytes transferred, which may be less than n. If there's a problem with the input parameters, -1 is returned.

The size, data, blen, and buf fields in the ring buffer must be valid or bad things will happen. The blen field gets adjusted to reflect the added data.

If the data transfer leaves the top of data within minfree of the top of the data buffer, a gap gets introduced, so the next read will begin at the bottom of the data buffer. 🫶

🔥 int rb_memcpy(RB_NEXUS *dst, RB_NEXUS *src, int n, int mode,int minfree);

Copy up to n bytes of data from one ring buffer to another.

⚠️ This MAY NOT be used to copy data from a buffer to itself!!! Like, genuinely, do not.

Returns the number of bytes which were moved, limited by either how many more will fit into dst or how many are available in src.

If mode is not zero, the src ring buffer is modified to remove the copied bytes by adjusting blen and data. The src data buffer itself is not modified.

If the data transfer leaves the top of data within minfree of the top of the data buffer, a gap gets introduced, so that the next read will begin at the bottom of the data buffer. 🌸

🧊 void rb_reset(RB_NEXUS *rb);

Effectively empty a ring buffer by setting blen, data, and gap to zero — but it does not clear the contents of the buffer. Ghost data stays ghosting. 👻

👑 RB_NEXUS *rb_shadow_set_create(RB_NEXUS *rb, int count);

Create a shadow set of a ring buffer. Okay so a shadow set is an array of ring buffers that have independent data, gap, blen, and size fields, but share the single data buffer of the ring buffer they shadow. Iconic, honestly.

Shadow sets let different parts of a program access the common data buffer without interfering with each other. One member can add new data to the common buffer while several others are simultaneously used to copy out existing data. Main character energy for the whole squad.

🚨 No shadow member may be used to both add and remove data from the buffer in a single operation. If you do that, there's no way to reconcile the changes. Periodt.

After one operation has been performed on each shadow set member, their independent data, gap, blen, and size fields must be reconciled with rb_shadow_set_reconcile before any further operations happen. If the size field of a shadow set member is set to 0, then that shadow set member gets ignored by rb_shadow_set_reconcile. You'd typically do this when a data stream has closed — since there'll be no further operations on that shadow set member, there's no need to reconcile its fields on each cycle. Bye bestie, you're free. 🦋

If rb doesn't exist or the shadow can't be created, NULL is returned.

🧚 void rb_shadow_set_delete(RB_NEXUS **rb);

Delete a ring buffer shadow set. All memory is released and the shadow set pointer rb is set to Null. The ring buffer from which the shadow set was made is not affected — but its contents are typically invalid, and rb_clear should be applied before it's used again. Clean slate required. 💅

💖 int rb_shadow_set_reconcile(RB_NEXUS *rb, int count, int input, int reference);

When you're operating on shadow sets in a given IO cycle, one read into the buffer and N writes from the buffer may be carried out. The read adds data to the "top" of the buffer, and the writes copy variable amounts of data from the bottom.

⚠️ It is essential to reconcile the shadow set before any shadow set member has an IO operation repeated on it. Failure to do so could result in the top of the buffer overwriting the bottom, or the top not being read because the buffer appeared "full" when it wasn't. Like, genuinely chaos. 😭

The number of members in the shadow set is count. The number of the member which added data is input. reference is the number of a member which was neither read nor written — it's used by rb_shadow_set_reconcile to determine what to change in all the other members. Sneaky little referee, iykyk.

The return value is -1 on failure, or the size of the member with the most data on success. 🏆

🌟 int rb_to_flat(char *array, RB_NEXUS *src, int n, int mode);

Copy n bytes of data from a ring buffer to a flat buffer. n may be more than the number of bytes stored in src, in which case the number of bytes present will be transferred. Returns the number of bytes transferred. If there's a problem with the input parameters, -1 is returned.

If mode is 0, the ring buffer is not modified. If nonzero, the data which is copied is effectively removed from the ring buffer by adjusting the data, blen, and gap fields. Choose your fighter. ⭐

🎀 int rb_to_flat_tail(char *array, RB_NEXUS *src, int n, int mode);

Exactly like rb_to_flat except the data is removed from the top of the data region instead of the bottom. Same vibe, opposite end. 🙈

💀 int rb_sanity(RB_NEXUS *rb, char *string);

Sanity check the fields in rb. This is primarily for debugging. Returns 0 if the data structure is ok. Returns 1 and emits diagnostic information followed by string if the data structure is invalid. Your bug-sniffing bestie. 🔍

✨ NIO LIBRARIES ✨

Okay bestie, so the files nio.c and nio.h are literally the whole vibe — they contain the nettee command and IO functions, plus all the NCMD symbols used in the examples. There are so many more functions available than what's described here (ngl, this is just the highlight reel). The top-level functions that the parent or child calls to enable the nettee command interface are described in the following sections 🫶

These definitions are available for simplifying access to certain information associated with the IO streams and their associated ring buffers. Most of these are used in the Example Child Program (iykyk):

NCMD_RB_SIZE(A)  (ncmd->rb[A].size)                 /* rb for stream A, total storage capacity  */
NCMD_RB_LEN(A)   (ncmd->rb[A].blen)                 /* rb for stream A, bytes of data stored    */
NCMD_RB_FREE(A)  (ncmd->rb[A].size - ncmd->rb[A].blen - ncmd->rb[A].gap)
                                                    /* rb for stream A, remaining capacity      */
NCMD_RB_CELL(A)  (&ncmd->rb[A])                     /* pointer to the rb for stream A           */
NCMD_IO_FD(A)    (ncmd->io_fd[A])                   /* the fd associated with stream  A         */
NCMD_LOOP_MASK   (ncmd->loop_mask)                  /* the loop mask                            */
NCMD_MINFREE     (ncmd->minfree)                    /* the minfree value                        */
NCMD_IO_TOTAL    (ncmd->num_total)                  /* the number of io streams                 */
NCMD_IS_IN(A)    (ncmd->io_type[A] & NCMD_FD_IN)    /* true for an input   stream               */
NCMD_IS_OUT(A)   (ncmd->io_type[A] & NCMD_FD_OUT)   /* true for an output  stream               */
NCMD_IS_NET(A)   (ncmd->io_type[A] & NCMD_FD_NET)   /* true for a  network stream               */
NCMD_IS_LOCAL(A) (ncmd->io_type[A] & NCMD_FD_LOCAL) /* true for a  local   stream               */
NCMD_DEBUG(B)    if(ncmd->debug){ B ; (void) fflush(stderr); }
                                                    /* conditionally emit a debug message       */

🫶 Overview of the Parent Interface 🫶

More complete details for how the parent controls the child are shown below in the Example Parent Program section. In brief (fr fr), the parent calls three primary functions. The first is ncmd_p_init, which creates the necessary data structure. The parent then opens whatever data streams the child will have access to and stores the information in this data structure. The parent sets up sockets to talk to the child and forks a process which runs the child. After the fork the parent runs ncmd_control_child, which handles communications with the child. Finally, after the child exits, the parent may call ncmd_p_free to release the data structure. The primary functions called by the parent are:

  • void ncmd_p_free(NCMD_P_NEXUS *ncmd); — Release the data associated with controlling the child. This is called after the child exits. (Cue the clean-up montage 🎀)
  • NCMD_P_NEXUS *ncmd_p_init(int io, int size, IOP_NEXUS *iop, int misses, int interval, int conmask); — Allocate a data structure which is used while controlling the child. The values for size, minfree and the net and lcl values for minwrite, maxread, and maxwrite are described in the -buf_size (etc.) sections of the nettee man page. The child process may be configured to send heartbeat messages 💓. The value of interval, which is a time in seconds, is passed to the child to set up this heartbeat. The parent looks for the heartbeat at the expected interval. If interval is zero the parent does not monitor and the child does not generate heartbeat messages. If the parent is monitoring and the expected heartbeat has not appeared in misses intervals, the parent assumes that the child has crashed, locked up, or is for some other reason broken. The parent then attempts to kill the child and then exits. (Not me being dramatic, but that's literally the protocol 😭)

The child may encounter read/write errors from/to local/network data streams. When it does so it will send a message to the parent:

X16X ERROR [READ|WRITE] stream

What the parent does at that point is determined by the conmask (ContinueON) mask. First the parent determines if the designated stream was Read or Write, Local or Network. The parent knows this because it kept track of that information when it opened that stream, before passing the information to the child. If the corresponding bit in conmask is not set the parent will send the child an EXIT message. However if the bit is set it will instead send an START message. In the former case the child will exit immediately; in the latter it will resume processing, albeit without the failed data stream. It is up to the programmer to determine for each type of child process whether or not to continue on these types of errors. The defined conmask bits are:

  • NCMD_ERR_CONWF 0x0010 — continue on network write failure
  • NCMD_ERR_COLWF 0x0020 — continue on local write failure
  • NCMD_ERR_CONRF 0x0040 — continue on network read failure
  • NCMD_ERR_COLRF 0x0080 — continue on local read failure

All other bits are reserved. Periodt.

  • int ncmd_control_child(NCMD_P_NEXUS *ncmd); — This is the function that keeps track of the child process. The first thing it does is send the following messages to the child, using data already stored in the ncmd data structure:
IOP size minfree net_nw net_xr net_xw lcl_nw lcl_xr lcl_xw
STATUS interval
STREAMS n                   ! number of IO streams
IO 0 name1 mask1            ! assign data stream to fd or file
IO 1 name2 mask2            ! assign data stream to fd or file
...
IO N nameN maskN
START                       ! go

During and after START it monitors responses. If an ERROR comes back it may send an EXIT to the child if the corresponding con_mask bit is not set.

Returns 0 on normal child exit. Returns with the NCMD_ERR status bit on abnormal exits. (So like, pay attention to that return value bestie 👀)

🌸 Overview of the Child Interface 🌸

More complete details for how the child program works are shown below in the Example Child Program section. In brief, the child calls three primary functions. The first is ncmd_c_init, which creates the necessary data structure. (Note that although the parent and child ncmd data structures share many field names they are actually different types: NCMD_P_NEXUS and NCMD_C_NEXUS — don't mix them up, bestie, it's giving "similar but not the same" 💅.) The child then repeatedly calls ncmd_all_io to obtain and send data. The child does no explicit IO on data streams but it may send an interface message directly to the parent. Between calls to this function it performs whatever processing that is required. The child controls when ncmd_all_io will return by setting the bits in the loop mask of the child's ncmd data structure. For instance:

NCMD_LOOP_MASK = (NCMD_DATA_IN_SOME | NCMD_CLOSED_IN_ALL | NCMD_DATA_OUT_LTMNW_SOME);

tells the ncmd_all_io function to not return until either there is some input data, all the inputs have closed, or at least one of the outputs contains less than minwrite bytes of data. The same bits are used to return complex status information via the function's return value. So the child can use:

if(ret & NCMD_DATA_IN_RD_SOME)

to see that new data is present in one or more input buffers. The complete set of bits is described in the Mask Bits section below. Finally, once all of the data has been read, processed, and sent off, the child releases the data structure with ncmd_c_free and then exits. (A little "thank u, next" moment ✨)

💅 Mask Bits 💅

Here are all of the mask bits used by the child program in its interactions with ncmd_all_io. These are either preset in the exit mask or returned in the status value:

  • NCMD_RUNNING — reading/writing data and commands, also processing
  • NCMD_DATA_IN_RD_NONE — data read on no open input streams
  • NCMD_DATA_IN_RD_SOME — data read on >=1 open input streams
  • NCMD_DATA_IN_RD_ALL — data read on all open input streams
  • NCMD_DATA_IN_NONE — data present >= 1 bytes on no input streams
  • NCMD_DATA_IN_SOME — data present >= 1 bytes on >=1 input streams
  • NCMD_DATA_IN_ALL — data present >= 1 bytes on all input streams
  • NCMD_DATA_OUT_WR_NONE — data written on no open output streams
  • NCMD_DATA_OUT_WR_SOME — data written on >=1 open output streams
  • NCMD_DATA_OUT_WR_ALL — data written on all open output streams
  • NCMD_DATA_OUT_NONE — data present on no open output streams
  • NCMD_DATA_OUT_SOME — data present on >=1 open output streams
  • NCMD_DATA_OUT_ALL — data present on all open output streams
  • NCMD_DATA_OUT_LTMNW_NONE — data present <minwrite bytes on no open output streams
  • NCMD_DATA_OUT_LTMNW_SOME — data present <minwrite bytes on >=1 open output streams
  • NCMD_DATA_OUT_LTMNW_ALL — data present <minwrite bytes on all open output streams
  • NCMD_CLOSED_IN_SOME — closed >=1 input streams
  • NCMD_CLOSED_IN_ALL — closed all input streams
  • NCMD_CLOSED_COMMAND — closed command stream

Special bits: (these may be set but are never returned as status — so like, they're the "set it and forget it" crew 🧚)

  • NCMD_FLUSH — Set minwrite to 1, so that all output buffers empty.
  • NCMD_CLEANUP — Emit DONE WRITE for any currently open output streams. Emit DONE READ for any currently open input streams. Emit DONE. SKIP the regular IO section (IO must have completed). Return an empty status mask.
  • NCMD_IGNORE_COMMAND — ignore commands from parent. (Not me ignoring the group chat 😭)

✨ Example child program ✨

Okay bestie this is the part where it all comes together — here's a whole actual working stub program that does the protocol thing so you can see what a real -process child looks like 👑 It's literally the reference implementation and the comments are also the documentation, iykyk 🫶

/*
Program:   stub_child_process.c
Version:   0.1
Date:      03-DEC-2007
Author:    David Mathog, Biology Division, Caltech
email:     mathog@caltech.edu
Copyright: 2007 David Mathog and California Institute of Technology (Caltech)
License:   GNU General Public License 2.
Description:

          This stub program implements the communications protocol needed to
          run as a -process in nettee.  See the comments in that
          program for the full syntax.

          This example just echoes all input to all outputs, monitors for
          command signals, and returns status signals.  If the input files
          are larger than the buffers their contents will be interleaved
          in more or less buffer sized chunks into the output.  This is a
          binary method so EOL in the input files has no special meaning.

          Compiles cleanly with:

          % gcc -Wall -pedantic -std=c99 -D_POSIX_SOURCE -o stub_child_process stub_child_process.c rb.c nio.c

Dependencies:
          rb.c    ring buffer routines
          nio.c   nettee cmd routines

Changes:
          V 0.1 03-DEC-2007, David Mathog <mathog@caltech.edu.  First version

*/
#include <stdlib.h>
#include <stdio.h>
#include <unistd.h>
#include <errno.h>
#include <string.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <sys/wait.h>
#include <sys/poll.h>
#include <netdb.h>
#include <sys/time.h>
#include <sys/mman.h>
#include <netinet/tcp.h>
#include <assert.h>
#include <ctype.h>
#include "rb.h"                   /* ring buffers   */
#include "nio.h"                  /* nettee io and ncmd pieces */

/* prototypes */

/* globals
NONE
*/

int main(int argc, char **argv){
/* Your variable declarations go here */
int i,j;
int ret;
int maxmove;
NCMD_C_NEXUS *ncmd=NULL;

          /* Your command line arguments are processed here */

          /* sets default values in the ncmd structure */
          ncmd=ncmd_c_init();
          ncmd->debug = 0; /* If 1, then emit NCMD_DEBUG messages */

          /*
          Here we set up a mask which determines when ncmd_all_io() will return, here the conditions are:
          NCMD_DATA_IN_SOME   = some input data is available, it may not have been read this cycle
          NCMD_CLOSED_IN_ALL  = the last input stream has closed
          NCMD_DATA_OUT_LTMNW_SOME  = at least one output buffers has been written to the greatest extent possible,
          (They may not be EMPTY if minwrite > 1!)

          Note that this example assumes that anything put in the output buffers will
          be written eventually.  That might not happen if minwrite was set larger than
          the size of one of the input files.  To handle cases like that one would test
          for conditions like

          (ret & NCMD_DATA_OUT_SOME) && (ret & NCMD_DATA_WR_NONE)
          */

          NCMD_LOOP_MASK = (NCMD_DATA_IN_SOME | NCMD_CLOSED_IN_ALL | NCMD_DATA_OUT_LTMNW_SOME);

          /* main loop here, this example just copies all input(s) to all output(s).  It copies
          only after one full set of writes, to all output streams, has completed.  */

          while(1){
          ret = ncmd_all_io(ncmd);
          if(ncmd->debug)ncmd_say_status("DEBUG child MAIN, after ncmd_all_io",ret);
          if(ret & NCMD_CLOSED_IN_ALL){
          NCMD_DEBUG( (void) fprintf(stderr,"DEBUG, in child MAIN all input closed\n"));
          if(ret & NCMD_DATA_IN_SOME){
          NCMD_DEBUG( (void) fprintf(stderr,"DEBUG, in child MAIN input data remains\n"));
          /* if all the inputs have closed we just need to finish moving whatever is already
          in the input buffers. Remove the condition for closed_in_all or the IO loop will
          not execute anymore since that status bit will be stuck "on". Set flush so that
          the output buffers will empty no matter how little data they contain (ie, ignore minwrite).
          Note that NCMD_FLUSH is not a status bit, so it can't ever prevent the IO loop from executing. */
          NCMD_LOOP_MASK = (NCMD_DATA_IN_SOME | NCMD_DATA_OUT_LTMNW_SOME | NCMD_FLUSH);
          }
          else if(ret & NCMD_DATA_OUT_SOME){
          NCMD_DEBUG( (void) fprintf(stderr,"DEBUG, in child MAIN output data remains to be flushed\n"));
          NCMD_LOOP_MASK = (NCMD_DATA_OUT_LTMNW_SOME | NCMD_FLUSH);
          }
          else {
          NCMD_DEBUG( (void) fprintf(stderr,"DEBUG, in child MAIN input data has been consumed\n"));
          /* We are done.  However...
          1. Use NCMD_CLEANUP to force the DONE WRITE values out and force status to 0.
          2. Command could still be open, use NCMD_IGNORE_COMMAND to force command to be ignored,
          otherwise ncmd_all_io will loop forever waiting for the next command.
          */
          NCMD_LOOP_MASK = NCMD_CLEANUP | NCMD_IGNORE_COMMAND;
          }
          }

          if((ret & NCMD_DATA_OUT_LTMNW_SOME) && (ret & NCMD_DATA_IN_SOME)){
          /* This is the place to put the code which actually does something with
          the IO Streams.  In this example for each pass where there is input
          data all of the output streams are scanned to find out the maximum
          number of bytes which can be moved to all of them.  Then that amount
          is transferred.
          */
          NCMD_DEBUG( (void) fprintf(stderr,"DEBUG, in child MAIN have input\n"));
          /* If most output data has been written and input data is available */
          /* There is input to work with and space in the output buffers are available, so copy at least some of it */
          for(j=0; j < NCMD_IO_TOTAL; j++){  /* over all possible inputs */
          /* This is an input and has data available.  It isn't necessary to check
          for a valid fd since no data could have been read on an invalid fd.  */
          if(NCMD_RB_LEN(j) && NCMD_IS_IN(j)){
          NCMD_DEBUG((void) fprintf(stderr,"DEBUG, in child MAIN SRC of data %d\n",j));
          maxmove=NCMD_RB_LEN(j);
          for(i=0; i< NCMD_IO_TOTAL; i++){
          if(NCMD_IO_FD(i) >= 0 && NCMD_IS_OUT(i)){
          if(NCMD_RB_FREE(i) < maxmove)maxmove=NCMD_RB_FREE(i);
          }
          }
          /* copy the data from this input buffer to each of the (open) output buffer(s).
          The buffers are the same size, so it will always fit.                     */
          for(i=0; i< NCMD_IO_TOTAL; i++){
          if(NCMD_IO_FD(i) >= 0 && NCMD_IS_OUT(i)){
          NCMD_DEBUG((void) fprintf(stderr,"DEBUG, in child MAIN DST of data %d\n",i));
          NCMD_DEBUG((void) fprintf(stderr,"DEBUG, in child MAIN move bytes: %8d\n",maxmove));
          ret = rb_memcpy(NCMD_RB_CELL(i), NCMD_RB_CELL(j), maxmove, RB_SRC_COPY, NCMD_MINFREE);
          if(ret != maxmove){
          send_status_exit("FAILED OTHER \"programming error, memcpy failed \"",
          NCMD_TOTAL_READ, 0, EXIT_FAILURE);
          }
          }
          }
          rb_adjust(NCMD_RB_CELL(j),maxmove);    /* throw out data in this input buffer */
          break;                    /* all output buffers were filled from this one input buffer */
          }
          }
          }
          else if(ret==0){
          /* We are completely done. */
          break;
          }
          /* there may be other ret values returned, but all just cause it to cycle back for another round */
          }
          NCMD_DEBUG((void) fprintf(stderr,"DEBUG child main, immediately before program exit\n"));
          ncmd_c_free(ncmd);
          exit(EXIT_SUCCESS);
}

Okay so let's do a little vibe check on what that program is actually doing, because reading 200 lines of C in one go is giving "wall of text" energy 😭

  • It pulls in rb.h (the ring buffers) and nio.h (the nettee io and ncmd pieces) — those two files are its whole personality.
  • It calls ncmd_c_init() to get a fresh NCMD_C_NEXUS *ncmd with all the defaults set, and sets ncmd->debug = 0 (flip it to 1 and you get NCMD_DEBUG messages, iconic for debugging ✨).
  • Then it sets NCMD_LOOP_MASK to control when ncmd_all_io() returns — the three conditions being NCMD_DATA_IN_SOME, NCMD_CLOSED_IN_ALL, and NCMD_DATA_OUT_LTMNW_SOME.
  • The loop calls ncmd_all_io(ncmd) forever, and depending on what came back it rewrites NCMD_LOOP_MASK:
    • all input closed but data remains → NCMD_DATA_IN_SOME | NCMD_DATA_OUT_LTMNW_SOME | NCMD_FLUSH (drop the closed bit or it'll be stuck "on" forever and the loop stops executing, bestie, that's the trap 🪤)
    • output data still to flush → NCMD_DATA_OUT_LTMNW_SOME | NCMD_FLUSH
    • nothing left → NCMD_CLEANUP | NCMD_IGNORE_COMMAND, so the DONE WRITE values get forced out with status 0, and the command gets ignored instead of ncmd_all_io waiting forever for a next command that's never coming 💀
  • When there's both input and output room, it scans every stream, computes maxmove as the smallest of the available input length and each open output's NCMD_RB_FREE(i), then rb_memcpy(..., RB_SRC_COPY, NCMD_MINFREE) from that one input cell into every open output cell, and rb_adjust()s the source to throw the copied bytes away. Then break, because one input buffer just filled everybody 🎀
  • If ret==0 it's done and it bails out of the loop, then ncmd_c_free(ncmd) and exit(EXIT_SUCCESS).
Heads up bestie: that comment about the minwrite assumption is a real caveat, not decoration — this example assumes anything you shove into the output buffers eventually gets written. If minwrite is bigger than one of your input files, that assumption falls apart and you'd need to test things like (ret & NCMD_DATA_OUT_SOME) && (ret & NCMD_DATA_WR_NONE) instead. Not me warning you in advance 🙈

Also NCMD_FLUSH is not a status bit — that's why it can never stop the IO loop from running. Little detail, huge consequences, fr fr 💅

And the other fun fact: because this is a binary method, EOL in the input files means nothing special, and if your input files are bigger than the buffers, their contents get interleaved into the output in more-or-less buffer-sized chunks. That's not a bug, that's just the vibe 🌸

More coming, don't go anywhere 👑

🌟 Example parent program 🌟

Bestie, before you run this sample program, do the prep work: create two text files, /tmp/input1.txt and /tmp/input2.txt, and put some data in them ✨. Then compile both the parent and child example programs and leave them in the same directory. This parent program emulates exactly what nettee does when it runs the child example program — so running the parent will create three output files: /tmp/output3.txt, /tmp/output4.txt, and /tmp/output5.txt 🎀.

/*
Program:   stub_parent_process.c
Version:   0.2
Date:      31-OCT-2008
Author:    David Mathog, Biology Division, Caltech
email:     mathog@caltech.edu
Copyright: 2008 David Mathog and California Institute of Technology (Caltech)
License:   GNU General Public License 2.
Description:

This is a stub showing how the parent program nettee interacts with the
child through the nettee cmd interface.  It may be used as a scaffold upon which
to build other programs which can call the same programs that netttee does.

gcc -Wall -pedantic -std=c99 -D_POSIX_SOURCE -o stub_parent_process stub_parent_process.c  nio.c rb.c

*/

#include <stdio.h>
#include <stdlib.h>
#include <unistd.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <sys/stat.h>
#include <fcntl.h>
#include "nio.h"

#define MAXSTREAMS   8
#define INPUTS       2
#define OUTPUTS      3
#define IO           (INPUTS + OUTPUTS)
#define MINFREE      NCMD_DEFAULT_MINFREE  /* minimum contiguous free space for a read, must be >= 1 !*/
#define MINWRITE     1  /* rb parameter, write even if only 1 byte of data in buffer */
#define MAXREAD   1024  /* maximum bytes read    per src before allowing other IO */
#define MAXWRITE  1024  /* maximum bytes written per dst before allowing other IO  */
#define SIZE      1024  /* rb parameter, buffer size */
#define MISSES       3  /* may miss status 3 times before error */
#define INTERVAL     2  /* 2 second status */
#define CONMASK    (NCMD_ERR_CONWF | NCMD_ERR_COLWF |  NCMD_ERR_CONRF | NCMD_ERR_COLRF) /* no errors are fatal, this is a pretty unusual setting... */

int main(void){
pid_t child;
int sv_down[2];              /*these are used for the socket*/
int sv_up[2];                /*these are used for the socket*/
char string[1000];
int i;
int status;
IOP_NEXUS iop;

NCMD_P_NEXUS *ncmd;

iop.size    = SIZE;
iop.minfree = MINFREE;
iop.net_maxread  = iop.lcl_maxread  = MAXREAD;
iop.net_minwrite = iop.lcl_minwrite = MINWRITE;
iop.net_maxwrite = iop.lcl_maxwrite = MAXWRITE;

ncmd = ncmd_p_init( IO, &iop,  MISSES, INTERVAL, CONMASK);
if(!ncmd)insane("fatal error: creating of parent ncmd structure failed");

/* Open all inputs, /tmp/input1.txt and /tmp/input2.txt, these should already have data in them */
for(i=0; i<INPUTS; i++){
sprintf(string,"/tmp/input%d.txt",i+1);
ncmd->io_fd[i]=open(string,O_RDONLY);
if(!ncmd->io_fd[i])insane("fatal error: open of an input failed");
ncmd->io_type[i] = NCMD_FD_IN + NCMD_FD_LOCAL;
}

/* Open all outputs, stdout and /tmp/output1/2/3.txt */
for(; i<IO; i++){
sprintf(string,"/tmp/output%d.txt",i+1);
ncmd->io_fd[i]=open(string,O_WRONLY | O_CREAT,0666);
if(!ncmd->io_fd[i])insane("fatal error: open of an output failed");
ncmd->io_type[i] = NCMD_FD_OUT + NCMD_FD_LOCAL;
}
for(i=0;i<IO; i++){ (void) fprintf(stderr,"DEBUG parent io_fd[%d] fd %d type %d\n",i,ncmd->io_fd[i],ncmd->io_type[i]); (void) fflush(stderr); }

sprintf(string,"./stub_child_process");
(void) fprintf(stderr,"pre fork, command is %s\n",string); (void) fflush(stderr);

if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv_down)) {insane("fatal error: socketpair down failed");}
if (socketpair(AF_UNIX, SOCK_STREAM, 0, sv_up)) {  insane("fatal error: socketpair up   failed");}
ncmd_p_preserve_std(ncmd);   /* dup stdin and/or stdout in either datain or dataout */

child=fork();
if (child > 0) {
ncmd_p_setchild(ncmd, child, sv_down[1], sv_up[0]);    /* write commands to sv_down[1] read from sv_up[0] */
(void) fprintf(stderr,"DEBUG i am PARENT \n"); (void) fflush(stderr);
status = ncmd_control_child(ncmd);
}
else {
(void) fprintf(stderr,"DEBUG i am CHILD with command: %s\n",string); (void) fflush(stderr);
dup2(sv_down[0], 0);    /* socket replaces stdin, read commands from here  */
dup2(sv_up[1], 1);      /* socket replaces stdout, write status to here    */
ncmd_set_nonblock(0);
ncmd_set_nonblock(1);
execlp("sh", "sh", "-c", string,NULL);
insane("fatal error: exec for cmd failed");
}
exit(EXIT_SUCCESS);
}

🦋 Example script for running the example child program 🦋

Okay so here's the vibe: it's usually easiest to debug the child program separately first, before you even think about using it with nettee -process 💅. That way, when something goes wrong, it goes wrong on a single node instead of — as Murphy dictates — every node in the distribution tree 😭💀. And because the interface is so simple, child programs may be started directly from shell scripts (like below) or even straight from the command line. Periodt.

In this example, three input files get created. The example child program is then fed a series of commands and runs. The three output files will consist of the contents of input2.txt concatenated to those of input1.txt — iconic behavior honestly 🌸. Here it's run within valgrind, so memory access problems may be detected 🔍. There's no IOP message here since this example uses the default buffer parameters. This script does essentially the same thing as the Example Parent Program, except the output files are output1.txt, output2.txt and output3.txt.

#!/bin/sh
rm -f /tmp/output1.txt
rm -f /tmp/output2.txt
rm -f /tmp/output3.txt
cat >/tmp/input1.txt <<EOD
from input 1 line 1
from input 1 line 2
from input 1 line 3
EOD
cat >/tmp/input2.txt <<EOD
from input 2 line 1
from input 2 line 2
from input 2 line 3
from input 2 line 4
EOD
valgrind --leak-check=full ./stub_child_process\
<<EOD \
>test_stdout.log\
2>test_stderr.log\
3</tmp/input1.txt\
4</tmp/input2.txt\
5>/tmp/output1.txt\
6>/tmp/output2.txt\
7>/tmp/output3.txt
STATUS 2
STREAMS 5
IO 0 3  5
IO 1 4  9
IO 2 5  6
IO 3 6  6
IO 4 7 10
START
EOD
# alternative form, by name
#./stub_child_process <<EOD >test_stdout.log 2>test_stderr.log
# STATUS 2
# SIZE 512
# STREAMS 5
# IO  0 /tmp/input1.txt    5
# IO  1 /tmp/input2.txt    9
# IO  2 /tmp/output1.txt   6
# IO  3 /tmp/output2.txt   6
# IO  4 /tmp/output3.txt  10
# START
EOD
👑 Heads up bestie: that second block is commented out on purpose — it's the "alternative form, by name" where you reference the files by name instead of by pre-assigned fd number. Flip the comments if you wanna try it 🎀

nettee is derived from Felix Rauch's dolly, which is available here: http://www.cs.inf.ethz.ch/CoPs/patagonia/#dolly

The nettee home page is: https://sourceforge.net/projects/nettee/

💖 COPYRIGHTS 💖

  • Copyright: 2019 David Mathog and Caltech.
  • Copyright: Felix Rauch and ETH Zurich

✨ LICENSE ✨

Freely distributed under the second GNU General Public License (GPL 2) 🫶

🎀 AUTHOR 🎀

David Mathog Biology Division, Caltech

And that's the whole page, bestie — no cap, you made it to the end 🥳💜 This was nettee(1), from nettee_cmd version 0.3.5, dated NOV 2019. Go forth and tee those bytes across the network, you absolute icon 🌟👑💅

📦 where this page comes from