paivana

stream_client.c (13190B)

---

 /*
   This file is part of paivana tests.
   Copyright (C) 2026 Taler Systems SA
 
   Paivana is free software; you can redistribute it and/or
   modify it under the terms of the GNU Affero General Public License
   as published by the Free Software Foundation; either version
   3, or (at your option) any later version.
 
   Paivana is distributed in the hope that it will be useful,
   but WITHOUT ANY WARRANTY; without even the implied warranty
   of MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See
   the GNU Affero General Public License for more details.
 
   You should have received a copy of the GNU Affero General Public
   License along with Paivana; see the file COPYING.  If not,
   write to the Free Software Foundation, Inc., 51 Franklin
   Street, Fifth Floor, Boston, MA 02110-1301, USA.
 */
 
 /**
  * @file stream_client.c
  * @brief HTTP client for the streaming tests: verifies a body against
  *        the generated pattern as it arrives, and reports what it saw
  *        about the *framing* rather than only about the bytes.
  *
  *        Incremental verification is the point.  A 200 MiB body
  *        written to a file and compared afterwards costs the disk and
  *        the wall clock of two extra copies, and — worse for these
  *        tests — says nothing about whether paivana streamed it or
  *        assembled it first.  Here nothing larger than one libcurl
  *        delivery is ever held, so the client cannot be the thing that
  *        blows up the memory bound the tests are checking.
  *
  *        What it prints on stdout, one `key=value' per line:
  *
  *          status=N          HTTP status
  *          bytes=N           body bytes received
  *          pattern=ok|CORRUPT|...  incremental pattern check
  *          chunked=yes|no    Transfer-Encoding: chunked on the wire
  *          content_length=N|none   the client-visible Content-Length
  *          curl=N            CURLcode (0 = success, 18 = partial)
  *          ttfb_ms=N         time to first body byte
  *          total_ms=N        time to last byte
  *          body=...          the response body, with --print-body
  *
  *        The rate options are what make the congestion cases
  *        reproducible.  Without a slow peer at one end, loopback and
  *        the kernel socket buffers absorb everything and no
  *        backpressure is ever exercised: paivana would look
  *        well-behaved whether or not it had any.
  *
  *        The exit status is 0 whenever the exchange was carried out,
  *        *including* when the transfer failed: which failure occurred
  *        is data the driver asserts on, not a reason to abort. Exit 1
  *        is reserved for the client itself being unable to run.
  *
  *        Usage:
  *          stream_client URL [--upload N] [--chunked-upload]
  *                            [--expect-bytes N] [--head] [--print-body]
  *                            [--abort-after N] [--read-rate N]
  *                            [--upload-rate N] [--recv-buffer N]
  */
 #include "platform.h"
 #include <curl/curl.h>
 #include <inttypes.h>
 #include <stdbool.h>
 #include <stdint.h>
 #include <stdio.h>
 #include <stdlib.h>
 #include <string.h>
 #include <strings.h>
 
 
 /**
  * The body byte at offset @a off.  Must agree with the same function
  * in `stream_upstream.c'.
  */
 static uint8_t
 pattern_at (uint64_t off)
 {
   return (uint8_t) ((off * 7) + (off >> 13));
 }
 
 
 /**
  * Body bytes received so far.
  */
 static uint64_t received;
 
 /**
  * Offset of the first byte that did not match the pattern, or
  * UINT64_MAX if they all have.
  */
 static uint64_t bad_at = UINT64_MAX;
 
 /**
  * Did the response carry `Transfer-Encoding: chunked'?
  */
 static bool saw_chunked;
 
 /**
  * `Content-Length' of the response, or UINT64_MAX if there was none.
  */
 static uint64_t saw_length = UINT64_MAX;
 
 /**
  * Bytes of request body still to send.
  */
 static uint64_t upload_left;
 
 /**
  * Offset of the next request-body byte to send.
  */
 static uint64_t upload_off;
 
 /**
  * CURLOPT_BUFFERSIZE, or 0 to leave libcurl's default.  A small value
  * makes libcurl read the socket in small units, so paivana's content
  * reader is called far more often and for far less each time --
  * hundreds of pause/resume transitions where the default buffer would
  * produce a handful.
  */
 static unsigned long recv_buffer;
 
 /**
  * Read the response body at no more than this many bytes per second,
  * or 0 for as fast as it arrives.
  */
 static uint64_t read_rate;
 
 /**
  * Send the request body at no more than this many bytes per second,
  * or 0 for as fast as libcurl will take it.
  */
 static uint64_t upload_rate;
 
 /**
  * Abort the transfer once this many body bytes have arrived, or 0
  * never to.  For the cases about what paivana does when its client
  * walks away mid-response.
  */
 static uint64_t abort_after;
 
 /**
  * Print the response body rather than checking it against the
  * pattern?  For the targets whose answer is a short report about the
  * request we sent, which is the interesting half of an upload case.
  */
 static bool print_body;
 
 /**
  * First #BODY_KEEP bytes of the response, when @e print_body is set.
  */
 static char body_buf[512];
 
 /**
  * Bytes held in @e body_buf.
  */
 static size_t body_len;
 
 /**
  * Time of the first body byte, or 0 if none has arrived.
  */
 static struct timespec first_byte;
 
 /**
  * When the request started.
  */
 static struct timespec started;
 
 
 static long
 elapsed_ms (const struct timespec *from,
             const struct timespec *to)
 {
   return (long) ((to->tv_sec - from->tv_sec) * 1000
                  + (to->tv_nsec - from->tv_nsec) / 1000000);
 }
 
 
 /**
  * Sleep for @a ms milliseconds, resuming across signals.
  */
 static void
 sleep_ms (long ms)
 {
   struct timespec ts;
 
   if (0 >= ms)
     return;
   ts.tv_sec = ms / 1000;
   ts.tv_nsec = (ms % 1000) * 1000000L;
   while ( (0 != nanosleep (&ts, &ts)) &&
           (EINTR == errno) )
     ; /* again */
 }
 
 
 /**
  * Hold @a moved bytes down to @a rate bytes per second, measuring from
  * @a since.
  *
  * Paced against the clock rather than by a fixed nap per callback: the
  * callback is handed whatever libcurl happened to read, so a fixed nap
  * would make the actual rate depend on the delivery size and the cases
  * that assert on elapsed time would be measuring the wrong thing.
  *
  * @param since when the transfer started
  * @param moved bytes moved so far
  * @param rate bytes per second to hold to
  */
 static void
 pace (const struct timespec *since,
       uint64_t moved,
       uint64_t rate)
 {
   struct timespec now;
   long due_ms;
   long spent_ms;
 
   if (0 == rate)
     return;
   clock_gettime (CLOCK_MONOTONIC,
                  &now);
   due_ms = (long) ((moved * 1000ULL) / rate);
   spent_ms = elapsed_ms (since,
                          &now);
   sleep_ms (due_ms - spent_ms);
 }
 
 
 static size_t
 write_cb (char *ptr,
           size_t size,
           size_t nmemb,
           void *cls)
 {
   size_t n = size * nmemb;
 
   (void) cls;
   if (0 == received)
     clock_gettime (CLOCK_MONOTONIC,
                    &first_byte);
   if (print_body)
   {
     size_t room = sizeof (body_buf) - body_len;
     size_t take = (n < room) ? n : room;
 
     memcpy (&body_buf[body_len],
             ptr,
             take);
     body_len += take;
   }
   else
   {
     for (size_t i = 0; i < n; i++)
       if ( (UINT64_MAX == bad_at) &&
            ((uint8_t) ptr[i] != pattern_at (received + i)) )
         bad_at = received + i;
   }
   received += n;
   if ( (0 != abort_after) &&
        (received >= abort_after) )
     return 0; /* fail the transfer: this is us hanging up */
   /* Blocking here is the whole point: libcurl stops reading the socket
      while we sleep, its receive window closes, and paivana's own
      download ring backs up.  That is the congestion the cases are
      about, and there is no way to produce it on loopback without it. */
   pace (&first_byte,
         received,
         read_rate);
   return n;
 }
 
 
 static size_t
 header_cb (char *ptr,
            size_t size,
            size_t nmemb,
            void *cls)
 {
   size_t n = size * nmemb;
   static const char te[] = "transfer-encoding:";
   static const char cl[] = "content-length:";
 
   (void) cls;
   if ( (n >= sizeof (te) - 1) &&
        (0 == strncasecmp (ptr, te, sizeof (te) - 1)) )
   {
     /* The whole point of looking is to see the framing paivana chose,
        so match the value rather than assume it. */
     for (size_t i = sizeof (te) - 1; i + 6 < n; i++)
       if (0 == strncasecmp (&ptr[i], "chunked", 7))
       {
         saw_chunked = true;
         break;
       }
   }
   if ( (n >= sizeof (cl) - 1) &&
        (0 == strncasecmp (ptr, cl, sizeof (cl) - 1)) )
     saw_length = strtoull (&ptr[sizeof (cl) - 1],
                            NULL,
                            10);
   return n;
 }
 
 
 static size_t
 read_cb (char *ptr,
          size_t size,
          size_t nmemb,
          void *cls)
 {
   size_t room = size * nmemb;
   size_t n;
 
   (void) cls;
   n = (size_t) ((upload_left < (uint64_t) room)
                 ? upload_left
                 : (uint64_t) room);
   for (size_t i = 0; i < n; i++)
     ptr[i] = (char) pattern_at (upload_off + i);
   upload_off += n;
   upload_left -= n;
   pace (&started,
         upload_off,
         upload_rate);
   return n;
 }
 
 
 int
 main (int argc,
       char **argv)
 {
   CURL *eh;
   CURLcode rc;
   long status = 0;
   const char *url = NULL;
   uint64_t upload = 0;
   bool chunked_upload = false;
   bool head = false;
   uint64_t expect = UINT64_MAX;
   struct timespec done;
   struct curl_slist *hdrs = NULL;
 
   for (int i = 1; i < argc; i++)
   {
     if (0 == strcmp (argv[i], "--upload"))
       upload = strtoull (argv[++i], NULL, 10);
     else if (0 == strcmp (argv[i], "--chunked-upload"))
       chunked_upload = true;
     else if (0 == strcmp (argv[i], "--head"))
       head = true;
     else if (0 == strcmp (argv[i], "--print-body"))
       print_body = true;
     else if (0 == strcmp (argv[i], "--abort-after"))
       abort_after = strtoull (argv[++i], NULL, 10);
     else if (0 == strcmp (argv[i], "--read-rate"))
       read_rate = strtoull (argv[++i], NULL, 10);
     else if (0 == strcmp (argv[i], "--upload-rate"))
       upload_rate = strtoull (argv[++i], NULL, 10);
     else if (0 == strcmp (argv[i], "--recv-buffer"))
       recv_buffer = strtoul (argv[++i], NULL, 10);
     else if (0 == strcmp (argv[i], "--expect-bytes"))
       expect = strtoull (argv[++i], NULL, 10);
     else if (NULL == url)
       url = argv[i];
     else
     {
       fprintf (stderr,
                "unexpected argument `%s'\n",
                argv[i]);
       return 1;
     }
   }
   if (NULL == url)
   {
     fprintf (stderr,
              "usage: %s URL [--upload N] [--chunked-upload]"
              " [--head] [--print-body] [--expect-bytes N]"
              " [--abort-after N] [--read-rate N] [--upload-rate N]"
              " [--recv-buffer N]\n",
              argv[0]);
     return 1;
   }
   eh = curl_easy_init ();
   if (NULL == eh)
   {
     fprintf (stderr,
              "curl_easy_init failed\n");
     return 1;
   }
   upload_left = upload;
   curl_easy_setopt (eh, CURLOPT_URL, url);
   curl_easy_setopt (eh, CURLOPT_WRITEFUNCTION, &write_cb);
   curl_easy_setopt (eh, CURLOPT_HEADERFUNCTION, &header_cb);
   curl_easy_setopt (eh, CURLOPT_NOSIGNAL, 1L);
   if (0 != recv_buffer)
     curl_easy_setopt (eh,
                       CURLOPT_BUFFERSIZE,
                       (long) recv_buffer);
   if (head)
   {
     curl_easy_setopt (eh, CURLOPT_NOBODY, 1L);
   }
   else if (0 != upload)
   {
     curl_easy_setopt (eh, CURLOPT_POST, 1L);
     curl_easy_setopt (eh, CURLOPT_READFUNCTION, &read_cb);
     if (chunked_upload)
     {
       /* Say nothing about the length and libcurl chunks; this is the
          framing paivana has to reproduce upstream. */
       hdrs = curl_slist_append (hdrs,
                                 "Transfer-Encoding: chunked");
       curl_easy_setopt (eh, CURLOPT_HTTPHEADER, hdrs);
     }
     else
     {
       curl_easy_setopt (eh,
                         CURLOPT_POSTFIELDSIZE_LARGE,
                         (curl_off_t) upload);
     }
   }
   clock_gettime (CLOCK_MONOTONIC, &started);
   rc = curl_easy_perform (eh);
   clock_gettime (CLOCK_MONOTONIC, &done);
   curl_easy_getinfo (eh,
                      CURLINFO_RESPONSE_CODE,
                      &status);
   printf ("status=%ld\n", status);
   printf ("bytes=%" PRIu64 "\n", received);
   if (print_body)
     printf ("pattern=n/a\n");
   else if (UINT64_MAX != bad_at)
     printf ("pattern=CORRUPT at %" PRIu64 "\n", bad_at);
   else if ( (UINT64_MAX != expect) &&
             (received != expect) )
     printf ("pattern=SHORT want %" PRIu64 "\n", expect);
   else
     printf ("pattern=ok\n");
   printf ("chunked=%s\n", saw_chunked ? "yes" : "no");
   if (UINT64_MAX == saw_length)
     printf ("content_length=none\n");
   else
     printf ("content_length=%" PRIu64 "\n", saw_length);
   printf ("curl=%d\n", (int) rc);
   printf ("ttfb_ms=%ld\n",
           (0 == received) ? -1 : elapsed_ms (&started, &first_byte));
   printf ("total_ms=%ld\n", elapsed_ms (&started, &done));
   if (print_body)
     printf ("body=%.*s\n",
             (int) body_len,
             body_buf);
   curl_slist_free_all (hdrs);
   curl_easy_cleanup (eh);
   return 0;
 }
 
 
 /* end of stream_client.c */