curl/lib/cw-pause.c

/***************************************************************************
 *                                  _   _ ____  _
 *  Project                     ___| | | |  _ \| |
 *                             / __| | | | |_) | |
 *                            | (__| |_| |  _ <| |___
 *                             \___|\___/|_| \_\_____|
 *
 * Copyright (C) Daniel Stenberg, <daniel@haxx.se>, et al.
 *
 * This software is licensed as described in the file COPYING, which
 * you should have received as part of this distribution. The terms
 * are also available at https://curl.se/docs/copyright.html.
 *
 * You may opt to use, copy, modify, merge, publish, distribute and/or sell
 * copies of the Software, and permit persons to whom the Software is
 * furnished to do so, under the terms of the COPYING file.
 *
 * This software is distributed on an "AS IS" basis, WITHOUT WARRANTY OF ANY
 * KIND, either express or implied.
 *
 * SPDX-License-Identifier: curl
 *
 ***************************************************************************/

#include "curl_setup.h"

#include <curl/curl.h>

#include "urldata.h"
#include "bufq.h"
#include "cfilters.h"
#include "headers.h"
#include "multiif.h"
#include "sendf.h"
#include "cw-pause.h"


/* body dynbuf sizes */
#define CW_PAUSE_BUF_CHUNK         (16 * 1024)
/* when content decoding, write data in chunks */
#define CW_PAUSE_DEC_WRITE_CHUNK   (4096)

struct cw_pause_buf {
  struct cw_pause_buf *next;
  struct bufq b;
  int type;
};

static struct cw_pause_buf *cw_pause_buf_create(int type, size_t buflen)
{
  struct cw_pause_buf *cwbuf = curlx_calloc(1, sizeof(*cwbuf));
  if(cwbuf) {
    cwbuf->type = type;
    if(type & CLIENTWRITE_BODY)
      Curl_bufq_init2(&cwbuf->b, CW_PAUSE_BUF_CHUNK, 1,
                      (BUFQ_OPT_SOFT_LIMIT|BUFQ_OPT_NO_SPARES));
    else
      Curl_bufq_init(&cwbuf->b, buflen, 1);
  }
  return cwbuf;
}

static void cw_pause_buf_free(struct cw_pause_buf *cwbuf)
{
  if(cwbuf) {
    Curl_bufq_free(&cwbuf->b);
    curlx_free(cwbuf);
  }
}

struct cw_pause_ctx {
  struct Curl_cwriter super;
  struct cw_pause_buf *buf;
  size_t buf_total;
};

static CURLcode cw_pause_write(struct Curl_easy *data,
                               struct Curl_cwriter *writer, int type,
                               const char *buf, size_t nbytes);
static void cw_pause_close(struct Curl_easy *data,
                           struct Curl_cwriter *writer);
static CURLcode cw_pause_init(struct Curl_easy *data,
                              struct Curl_cwriter *writer);

const struct Curl_cwtype Curl_cwt_pause = {
  "cw-pause",
  NULL,
  cw_pause_init,
  cw_pause_write,
  cw_pause_close,
  sizeof(struct cw_pause_ctx)
};

static CURLcode cw_pause_init(struct Curl_easy *data,
                              struct Curl_cwriter *writer)
{
  struct cw_pause_ctx *ctx = writer->ctx;
  (void)data;
  ctx->buf = NULL;
  return CURLE_OK;
}

static void cw_pause_bufs_free(struct cw_pause_ctx *ctx)
{
  while(ctx->buf) {
    struct cw_pause_buf *next = ctx->buf->next;
    cw_pause_buf_free(ctx->buf);
    ctx->buf = next;
  }
}

static void cw_pause_close(struct Curl_easy *data, struct Curl_cwriter *writer)
{
  struct cw_pause_ctx *ctx = writer->ctx;

  (void)data;
  cw_pause_bufs_free(ctx);
}

static CURLcode cw_pause_flush(struct Curl_easy *data,
                               struct Curl_cwriter *cw_pause)
{
  struct cw_pause_ctx *ctx = (struct cw_pause_ctx *)cw_pause;
  bool decoding = Curl_cwriter_is_content_decoding(data);
  CURLcode result = CURLE_OK;

  /* write the end of the chain until it blocks or gets empty */
  while(ctx->buf && !Curl_cwriter_is_paused(data)) {
    struct cw_pause_buf **plast = &ctx->buf;
    size_t blen, wlen = 0;
    const unsigned char *buf = NULL;

    while((*plast)->next) /* got to last in list */
      plast = &(*plast)->next;
    if(Curl_bufq_peek(&(*plast)->b, &buf, &blen)) {
      wlen = (decoding && ((*plast)->type & CLIENTWRITE_BODY)) ?
             CURLMIN(blen, CW_PAUSE_DEC_WRITE_CHUNK) : blen;
      result = Curl_cwriter_write(data, cw_pause->next, (*plast)->type,
                                  (const char *)buf, wlen);
      CURL_TRC_WRITE(data, "[PAUSE] flushed %zu/%zu bytes, type=%x -> %d",
                     wlen, ctx->buf_total, (*plast)->type, result);
      Curl_bufq_skip(&(*plast)->b, wlen);
      DEBUGASSERT(ctx->buf_total >= wlen);
      ctx->buf_total -= wlen;
      if(result)
        return result;
    }
    else if((*plast)->type & CLIENTWRITE_EOS) {
      result = Curl_cwriter_write(data, cw_pause->next, (*plast)->type,
                                  (const char *)buf, 0);
      CURL_TRC_WRITE(data, "[PAUSE] flushed 0/%zu bytes, type=%x -> %d",
                     ctx->buf_total, (*plast)->type, result);
    }

    if(Curl_bufq_is_empty(&(*plast)->b)) {
      cw_pause_buf_free(*plast);
      *plast = NULL;
    }
  }
  return result;
}

static CURLcode cw_pause_write(struct Curl_easy *data,
                               struct Curl_cwriter *writer, int type,
                               const char *buf, size_t blen)
{
  struct cw_pause_ctx *ctx = writer->ctx;
  CURLcode result = CURLE_OK;
  size_t wlen = 0;
  bool decoding = Curl_cwriter_is_content_decoding(data);

  if(ctx->buf && !Curl_cwriter_is_paused(data)) {
    result = cw_pause_flush(data, writer);
    if(result)
      return result;
  }

  while(!ctx->buf && !Curl_cwriter_is_paused(data)) {
    int wtype = type;
    DEBUGASSERT(!ctx->buf);
    /* content decoding might blow up size considerably, write smaller
     * chunks to make pausing need buffer less. */
    wlen = (decoding && (type & CLIENTWRITE_BODY)) ?
           CURLMIN(blen, CW_PAUSE_DEC_WRITE_CHUNK) : blen;
    if(wlen < blen)
      wtype &= ~CLIENTWRITE_EOS;
    result = Curl_cwriter_write(data, writer->next, wtype, buf, wlen);
    CURL_TRC_WRITE(data, "[PAUSE] writing %zu/%zu bytes of type %x -> %d",
                   wlen, blen, wtype, result);
    if(result)
      return result;
    buf += wlen;
    blen -= wlen;
    if(!blen)
      return result;
  }

  do {
    size_t nwritten = 0;
    if(ctx->buf && (ctx->buf->type == type) && (type & CLIENTWRITE_BODY)) {
      /* same type and body, append to current buffer which has a soft
       * limit and should take everything up to OOM. */
      result = Curl_bufq_cwrite(&ctx->buf->b, buf, blen, &nwritten);
    }
    else {
      /* Need a new buf, type changed */
      struct cw_pause_buf *cwbuf = cw_pause_buf_create(type, blen);
      if(!cwbuf)
        return CURLE_OUT_OF_MEMORY;
      cwbuf->next = ctx->buf;
      ctx->buf = cwbuf;
      result = Curl_bufq_cwrite(&ctx->buf->b, buf, blen, &nwritten);
    }
    CURL_TRC_WRITE(data, "[PAUSE] buffer %zu more bytes of type %x, "
                   "total=%zu -> %d", nwritten, type, ctx->buf_total + wlen,
                   result);
    if(result)
      return result;
    buf += nwritten;
    blen -= nwritten;
    ctx->buf_total += nwritten;
  } while(blen);

  return result;
}

CURLcode Curl_cw_pause_flush(struct Curl_easy *data)
{
  struct Curl_cwriter *cw_pause;
  CURLcode result = CURLE_OK;

  cw_pause = Curl_cwriter_get_by_type(data, &Curl_cwt_pause);
  if(cw_pause)
    result = cw_pause_flush(data, cw_pause);

  return result;
}