net_http/webdav: stream the upload body from the file instead of reading it whole

A WebDAV upload read the complete file into a buffer, and
net_http_connection_set_content() then copied that buffer again, so
two copies of the file sat in memory for the whole transfer. Saves are
small, but a save state is not, and several uploads run at once.

net_http gains a request-body source: net_http_connection_set_content_source()
takes a Content-Length, a read callback and an optional rewind
callback. net_http_send_request() pulls the body from the source in
64 KiB runs and sends each as it arrives, so one run is all that is
ever held; a source that delivers fewer bytes than the advertised
length fails the request as a transport error would, since the header
has already promised them. When a request has to be replayed on a
fresh connection the source is rewound first; a source with no rewind,
or one whose rewind fails, is not replayed.

task_push_webdav_put_stream() wraps that for the task queue, and the
cloud-sync WebDAV driver uploads straight from the open RFILE through
filestream_read, rewinding with filestream_seek. The existing seek to
the start before the upload is kept, as the Digest-challenge retry
still reaches this path with the file at its end.

Checked with a 3 MB body PUT to a local server: the received body is
byte-identical (SHA-256) and the request completes with 201.

libretro-common/include/net/net_http.h

@@ -94,6 +94,41 @@ void net_http_connection_set_sink(struct http_connection_t *conn,
 void net_http_connection_set_content(struct http_connection_t *conn, const char *content_type,
       size_t content_length, const void *content);
 
+/**
+ * net_http_source_t:
+ *
+ * Called for the next run of request-body bytes: fill up to @len bytes
+ * of @buf and return how many were filled. Return 0 at the end of the
+ * body and a negative value on error; either one short of the
+ * advertised Content-Length fails the request like a transport error.
+ * Runs on whichever thread drives net_http_update().
+ **/
+typedef int64_t (*net_http_source_t)(void *userdata, void *buf, size_t len);
+
+/**
+ * net_http_source_rewind_t:
+ *
+ * Asked to restart the body from its first byte when the request has
+ * to be replayed on a fresh connection (a pooled connection the peer
+ * closed while idle). Return false if that is not possible; the
+ * request then fails instead of being replayed.
+ **/
+typedef bool (*net_http_source_rewind_t)(void *userdata);
+
+/**
+ * net_http_connection_set_content_source:
+ *
+ * Send a request body of @content_length bytes pulled from @source as
+ * the socket takes them, so peak memory is one send buffer rather than
+ * the whole body. Replaces any body given by
+ * net_http_connection_set_content(). @rewind may be NULL, in which
+ * case the request is never replayed on a fresh connection.
+ **/
+void net_http_connection_set_content_source(struct http_connection_t *conn,
+      const char *content_type, size_t content_length,
+      net_http_source_t source, net_http_source_rewind_t rewind,
+      void *userdata);
+
 const char *net_http_connection_url(struct http_connection_t *conn);
 
 const char* net_http_connection_method(struct http_connection_t* conn);

libretro-common/net/net_http.c

@@ -165,6 +165,11 @@ typedef struct request
    char *useragent;
    char *headers;
    size_t contentlength;
+   /* Streamed body: pulled from source in runs as the socket takes
+    * them, instead of being held whole in postdata. */
+   net_http_source_t source;
+   net_http_source_rewind_t source_rewind;
+   void *source_data;
    int port;
 } request_t;
 
@@ -214,6 +219,9 @@ struct http_connection_t
    size_t contentlength; /* ptr alignment */
    net_http_sink_t sink;
    void *sink_data;
+   net_http_source_t source;
+   net_http_source_rewind_t source_rewind;
+   void *source_data;
    int port;
    bool ssl;
 };
@@ -665,6 +673,9 @@ void net_http_connection_set_content(
 
    conn->contenttype   = content_type ? strdup(content_type) : NULL;
    conn->contentlength = content_length;
+   conn->source        = NULL;
+   conn->source_rewind = NULL;
+   conn->source_data   = NULL;
    if (content_length)
    {
       conn->postdata = malloc(content_length);
@@ -680,6 +691,23 @@ void net_http_connection_set_content(
    }
 }
 
+void net_http_connection_set_content_source(struct http_connection_t *conn,
+      const char *content_type, size_t content_length,
+      net_http_source_t source, net_http_source_rewind_t rewind,
+      void *userdata)
+{
+   if (conn->contenttype)
+      free(conn->contenttype);
+   if (conn->postdata)
+      free(conn->postdata);
+   conn->postdata      = NULL;
+   conn->contenttype   = content_type ? strdup(content_type) : NULL;
+   conn->contentlength = content_length;
+   conn->source        = source;
+   conn->source_rewind = rewind;
+   conn->source_data   = userdata;
+}
+
 const char *net_http_connection_url(struct http_connection_t *conn)
 {
    return conn->url;
@@ -1014,6 +1042,9 @@ struct http_t *net_http_new(struct http_connection_t *conn)
       conn->postdata            = NULL;
       conn->contentlength       = 0;
    }
+   state->request.source        = conn->source;
+   state->request.source_rewind = conn->source_rewind;
+   state->request.source_data   = conn->source_data;
    state->request.useragent= conn->useragent ? strdup(conn->useragent) : NULL;
    state->request.headers  = conn->headers ? strdup(conn->headers) : NULL;
    state->request.port     = conn->port;
@@ -1315,6 +1346,15 @@ static bool net_http_retry_fresh(struct http_t *state)
    if (!state->conn_reused || state->retried || state->response.pos)
       return false;
 
+   /* A streamed body may already be partly consumed; it has to be
+    * restarted from its first byte, or the request cannot be replayed. */
+   if (state->request.source)
+   {
+      if (     !state->request.source_rewind
+            || !state->request.source_rewind(state->request.source_data))
+         return false;
+   }
+
    net_http_log_transport_state(state, "retry_on_fresh_connection", -1);
 
    if (state->conn)
@@ -1362,6 +1402,47 @@ static void net_http_send_str(
    }
 }
 
+/* The body, pulled from the source one run at a time and sent as each
+ * run arrives. The send is blocking like the rest of the request, so
+ * one buffer is all that is ever held. The source has to deliver
+ * exactly Content-Length bytes: the header has already promised them,
+ * and a body that stops short would leave the server waiting. */
+#define NET_HTTP_SOURCE_RUN (64 * 1024)
+
+static void net_http_send_source(struct http_t *state)
+{
+   struct request *request = (struct request*)&state->request;
+   size_t sent  = 0;
+   size_t run   = request->contentlength < NET_HTTP_SOURCE_RUN
+      ? request->contentlength : NET_HTTP_SOURCE_RUN;
+   char  *buf   = (char*)malloc(run);
+
+   if (!buf)
+   {
+      state->err = true;
+      net_http_log_transport_state(state, "source_oom", -1);
+      return;
+   }
+
+   while (!state->err && sent < request->contentlength)
+   {
+      size_t  want = request->contentlength - sent;
+      int64_t got;
+      if (want > run)
+         want = run;
+      got = request->source(request->source_data, buf, want);
+      if (got <= 0)
+      {
+         state->err = true;
+         net_http_log_transport_state(state, "source_short", -1);
+         break;
+      }
+      net_http_send_str(state, buf, (size_t)got);
+      sent += (size_t)got;
+   }
+   free(buf);
+}
+
 static bool net_http_send_request(struct http_t *state)
 {
    struct request *request = (struct request*)&state->request;
@@ -1370,6 +1451,7 @@ static bool net_http_send_request(struct http_t *state)
          && request->method[0] == 'P'
          && request->method[1] == 'O' /* POST, not PUT */
          && !request->postdata
+         && !request->source
          && request->contentlength > 0)
    {
       state->err = true;
@@ -1448,7 +1530,9 @@ static bool net_http_send_request(struct http_t *state)
       net_http_send_str(state, "libretro", sizeof("libretro")-1);
    net_http_send_str(state, "\r\n", sizeof("\r\n")-1);
    net_http_send_str(state, "\r\n", sizeof("\r\n")-1);
-   if (request->postdata && request->contentlength)
+   if (request->source && request->contentlength)
+      net_http_send_source(state);
+   else if (request->postdata && request->contentlength)
       net_http_send_str(state, (const char*)request->postdata,
             request->contentlength);
    state->request_sent = true;

network/cloud_sync/webdav.c

@@ -1100,10 +1100,21 @@ static void webdav_update_cb(retro_task_t *task, void *task_data,
    free(webdav_cb_st);
 }
 
+/* The upload body, pulled from the open save file by net_http on the
+ * task thread as the socket takes it. */
+static int64_t webdav_upload_source(void *userdata, void *buf, size_t len)
+{
+   return filestream_read((RFILE*)userdata, buf, (int64_t)len);
+}
+
+static bool webdav_upload_rewind(void *userdata)
+{
+   return filestream_seek((RFILE*)userdata, 0, SEEK_SET) == 0;
+}
+
 static void webdav_do_update(bool success, webdav_cb_state_t *webdav_cb_st)
 {
    char            url_encoded[PATH_MAX_LENGTH];
-   void           *buf;
    int64_t         len;
    char           *auth_header;
 
@@ -1125,20 +1136,15 @@ static void webdav_do_update(bool success, webdav_cb_state_t *webdav_cb_st)
       return;
    }
 
-   /* TODO: would be better to read file as it's being written to wire, this is very inefficient */
-   /* Rewind first: a retry after a Digest challenge comes back here
-    * with the file already read to its end, and without the seek it
-    * read nothing and uploaded a buffer of uninitialised memory in
-    * place of the save.  A short read fails the upload for the same
-    * reason. */
+   /* The file is streamed onto the wire from rfile as the socket takes
+    * it, one send buffer at a time, rather than read whole into memory
+    * first. Rewind first: a retry after a Digest challenge comes back
+    * here with the file already read to its end, and without the seek
+    * the upload would carry nothing. A short read fails the upload. */
    len = filestream_get_size(webdav_cb_st->rfile);
-   buf = (len >= 0) ? malloc((size_t)(len + 1)) : NULL;
-   if (   !buf
-       || filestream_seek(webdav_cb_st->rfile, 0, SEEK_SET) < 0
-       || filestream_read(webdav_cb_st->rfile, buf, len) != len)
+   if (len < 0 || filestream_seek(webdav_cb_st->rfile, 0, SEEK_SET) < 0)
    {
       RARCH_ERR("[webdav] Could not read %s for upload.\n", webdav_cb_st->path);
-      free(buf);
       webdav_cb_st->cb(webdav_cb_st->user_data, webdav_cb_st->path, false, webdav_cb_st->rfile);
       free(webdav_cb_st);
       return;
@@ -1146,10 +1152,10 @@ static void webdav_do_update(bool success, webdav_cb_state_t *webdav_cb_st)
 
    RARCH_DBG("[webdav] PUT %s\n", url_encoded);
    auth_header = webdav_get_auth_header("PUT", url_encoded);
-   task_push_webdav_put(url_encoded, buf, len, true, auth_header, webdav_update_cb, webdav_cb_st);
+   task_push_webdav_put_stream(url_encoded,
+         webdav_upload_source, webdav_upload_rewind, webdav_cb_st->rfile,
+         (size_t)len, true, auth_header, webdav_update_cb, webdav_cb_st);
    free(auth_header);
-
-   free(buf);
 }
 
 /* Where the backup of @path goes: deleted/<path>-<yymmdd-hhmmss>, the

tasks/task_http.c

@@ -594,6 +594,34 @@ void* task_push_webdav_put(const char *url,
    return task_push_http_transfer_generic(conn, url, mute, false, cb, user_data);
 }
 
+void* task_push_webdav_put_stream(const char *url,
+      net_http_source_t source, net_http_source_rewind_t rewind,
+      void *source_data, size_t len, bool mute,
+      const char *headers, retro_task_callback_t cb, void *user_data)
+{
+   struct http_connection_t *conn;
+   char                      expect[1024]; /* TODO/FIXME - check size */
+   size_t                    _len;
+
+   if (!url || !*url || !source)
+      return NULL;
+
+   if (!(conn = net_http_connection_new(url, "PUT", NULL)))
+      return NULL;
+
+   _len = strlcpy_lit(expect, "Expect: 100-continue\r\n", sizeof(expect));
+   if (headers)
+   {
+      strlcpy(expect + _len, headers, sizeof(expect) - _len);
+      net_http_connection_set_headers(conn, expect);
+   }
+
+   net_http_connection_set_content_source(conn, NULL, len,
+         source, rewind, source_data);
+
+   return task_push_http_transfer_generic(conn, url, mute, false, cb, user_data);
+}
+
 void* task_push_webdav_delete(const char *url, bool mute,
       const char *headers,
       retro_task_callback_t cb, void *user_data)

tasks/tasks_internal.h

@@ -96,6 +96,7 @@ bool task_nbio_slice_within_budget(void *ud, size_t avail, size_t len);
 void task_window_progress_cb(retro_task_t *task);
 
 #ifdef HAVE_NETWORKING
+#include <net/net_http.h>
 typedef struct
 {
    char *data;
@@ -133,6 +134,13 @@ void *task_push_webdav_mkdir(const char *url, bool mute, const char *headers,
       retro_task_callback_t cb, void *userdata);
 void *task_push_webdav_put(const char *url, const void *put_data, size_t len, bool mute, const char *headers,
       retro_task_callback_t cb, void *userdata);
+/* PUT a body of @len bytes pulled from @source as the socket takes
+ * them, holding one send buffer rather than the whole file. @rewind
+ * restarts the body for a replay on a fresh connection; NULL means the
+ * request is never replayed. */
+void *task_push_webdav_put_stream(const char *url, net_http_source_t source,
+      net_http_source_rewind_t rewind, void *source_data, size_t len, bool mute,
+      const char *headers, retro_task_callback_t cb, void *user_data);
 void *task_push_webdav_delete(const char *url, bool mute, const char *headers,
       retro_task_callback_t cb, void *userdata);
 void *task_push_webdav_move(const char *url, const char *dest, bool mute, const char *headers,