Add nghttp2_send_data_callback to send DATA payload without copying

To avoid buffer copy in nghttp2_data_source_read_callback, this commit
introduces NGHTTP2_DATA_FLAG_NO_COPY and nghttp2_send_data_callback.
By using NGHTTP2_DATA_FLAG_NO_COPY in
nghttp2_data_source_read_callback, application can avoid to copy
application data to given buffer.  Instead, application has to
implement nghttp2_send_data_callback to send complete DATA frame by
itself.  We see noticeable performance increase in nghttpd and
tiny-nghttpd using this new feature.  On the other hand, nghttpx does
not show such difference, probably because buffer copy is not
bottleneck.  Using nghttp2_send_data_callback adds complexity, so it
is recommended to measure the performance to see whether this extra
complexity worth it.

doc/Makefile.am

@@ -71,6 +71,7 @@ APIDOCS= \
 	nghttp2_session_callbacks_set_recv_callback.rst \
 	nghttp2_session_callbacks_set_select_padding_callback.rst \
 	nghttp2_session_callbacks_set_send_callback.rst \
+	nghttp2_session_callbacks_set_send_data_callback.rst \
 	nghttp2_session_client_new.rst \
 	nghttp2_session_client_new2.rst \
 	nghttp2_session_client_new3.rst \

examples/tiny-nghttpd.c

@@ -658,19 +658,57 @@ static void stream_error(connection *conn, int32_t stream_id,
                             error_code);
 }
 
-static ssize_t fd_read_callback(nghttp2_session *session _U_,
+static int send_data_callback(nghttp2_session *session _U_,
-                                int32_t stream_id _U_, uint8_t *buf,
+                              nghttp2_frame *frame, const uint8_t *framehd,
-                                size_t length, uint32_t *data_flags,
+                              size_t length, nghttp2_data_source *source _U_,
-                                nghttp2_data_source *source,
+                              void *user_data) {
-                                void *user_data _U_) {
+  connection *conn = user_data;
-  stream *strm = source->ptr;
+  uint8_t *p = conn->buf.last;
-  ssize_t nread;
+  stream *strm;
 
-  while ((nread = read(strm->filefd, buf, length)) == -1 && errno == EINTR)
+  strm = nghttp2_session_get_stream_user_data(session, frame->hd.stream_id);
+
+  if (!strm) {
+    return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+  }
+
+  /* We never use padding in this program */
+  assert(frame->data.padlen == 0);
+
+  if ((size_t)io_buf_left(&conn->buf) < 9 + frame->hd.length) {
+    return NGHTTP2_ERR_WOULDBLOCK;
+  }
+
+  memcpy(p, framehd, 9);
+  p += 9;
+
+  while (length) {
+    ssize_t nread;
+    while ((nread = read(strm->filefd, p, length)) == -1 && errno == EINTR)
       ;
     if (nread == -1) {
       return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
     }
+
+    length -= nread;
+    p += nread;
+  }
+
+  conn->buf.last = p;
+
+  return 0;
+}
+
+static ssize_t fd_read_callback(nghttp2_session *session _U_,
+                                int32_t stream_id _U_, uint8_t *buf _U_,
+                                size_t length, uint32_t *data_flags,
+                                nghttp2_data_source *source,
+                                void *user_data _U_) {
+  stream *strm = source->ptr;
+  ssize_t nread = (int64_t)length < strm->fileleft ? length : strm->fileleft;
+
+  *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY;
+
   strm->fileleft -= nread;
   if (nread == 0 || strm->fileleft == 0) {
     if (strm->fileleft != 0) {
@@ -1274,6 +1312,8 @@ int main(int argc, char **argv) {
       shared_callbacks, on_stream_close_callback);
   nghttp2_session_callbacks_set_on_frame_not_send_callback(
       shared_callbacks, on_frame_not_send_callback);
+  nghttp2_session_callbacks_set_send_data_callback(shared_callbacks,
+                                                   send_data_callback);
 
   rv = nghttp2_option_new(&shared_option);
   if (rv != 0) {

lib/includes/nghttp2/nghttp2.h

@@ -711,7 +711,12 @@ typedef enum {
    * trailer header fields with `nghttp2_submit_request()` or
    * `nghttp2_submit_response()`.
    */
-  NGHTTP2_DATA_FLAG_NO_END_STREAM = 0x02
+  NGHTTP2_DATA_FLAG_NO_END_STREAM = 0x02,
+  /**
+   * Indicates that application will send complete DATA frame in
+   * :type:`nghttp2_send_data_callback`.
+   */
+  NGHTTP2_DATA_FLAG_NO_COPY = 0x04
 } nghttp2_data_flag;
 
 /**
@@ -724,6 +729,15 @@ typedef enum {
  * them in |buf| and return number of data stored in |buf|.  If EOF is
  * reached, set :enum:`NGHTTP2_DATA_FLAG_EOF` flag in |*data_flags|.
  *
+ * Sometime it is desirable to avoid copying data into |buf| and let
+ * application to send data directly.  To achieve this, set
+ * :enum:`NGHTTP2_DATA_FLAG_NO_COPY` to |*data_flags| (and possibly
+ * other flags, just like when we do copy), and return the number of
+ * bytes to send without copying data into |buf|.  The library, seeing
+ * :enum:`NGHTTP2_DATA_FLAG_NO_COPY`, will invoke
+ * :type:`nghttp2_send_data_callback`.  The application must send
+ * complete DATA frame in that callback.
+ *
  * If this callback is set by `nghttp2_submit_request()`,
  * `nghttp2_submit_response()` or `nghttp2_submit_headers()` and
  * `nghttp2_submit_data()` with flag parameter
@@ -1189,6 +1203,47 @@ typedef ssize_t (*nghttp2_send_callback)(nghttp2_session *session,
                                          const uint8_t *data, size_t length,
                                          int flags, void *user_data);
 
+/**
+ * @functypedef
+ *
+ * Callback function invoked when :enum:`NGHTTP2_DATA_FLAG_NO_COPY` is
+ * used in :type:`nghttp2_data_source_read_callback` to send complete
+ * DATA frame.
+ *
+ * The |frame| is a DATA frame to send.  The |framehd| is the
+ * serialized frame header (9 bytes). The |length| is the length of
+ * application data to send (this does not include padding).  The
+ * |source| is the same pointer passed to
+ * :type:`nghttp2_data_source_read_callback`.
+ *
+ * The application first must send frame header |framehd| of length 9
+ * bytes.  If ``frame->padlen > 0``, send 1 byte of value
+ * ``frame->padlen - 1``.  Then send exactly |length| bytes of
+ * application data.  Finally, if ``frame->padlen > 0``, send
+ * ``frame->padlen - 1`` bytes of zero (they are padding).
+ *
+ * The application has to send complete DATA frame in this callback.
+ * If all data were written successfully, return 0.
+ *
+ * If it cannot send it all, just return
+ * :enum:`NGHTTP2_ERR_WOULDBLOCK`; the library will call this callback
+ * with the same parameters later (It is recommended to send complete
+ * DATA frame at once in this function to deal with error; if partial
+ * frame data has already sent, it is impossible to send another data
+ * in that state, and all we can do is tear down connection).  If
+ * application decided to reset this stream, return
+ * :enum:`NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE`, then the library
+ * will send RST_STREAM with INTERNAL_ERROR as error code.  The
+ * application can also return :enum:`NGHTTP2_ERR_CALLBACK_FAILURE`,
+ * which will result in connection closure.  Returning any other value
+ * is treated as :enum:`NGHTTP2_ERR_CALLBACK_FAILURE` is returned.
+ */
+typedef int (*nghttp2_send_data_callback)(nghttp2_session *session,
+                                          nghttp2_frame *frame,
+                                          const uint8_t *framehd, size_t length,
+                                          nghttp2_data_source *source,
+                                          void *user_data);
+
 /**
  * @functypedef
  *
@@ -1787,6 +1842,17 @@ NGHTTP2_EXTERN void nghttp2_session_callbacks_set_on_begin_frame_callback(
     nghttp2_session_callbacks *cbs,
     nghttp2_on_begin_frame_callback on_begin_frame_callback);
 
+/**
+ * @function
+ *
+ * Sets callback function invoked when
+ * :enum:`NGHTTP2_DATA_FLAG_NO_COPY` is used in
+ * :type:`nghttp2_data_source_read_callback` to avoid data copy.
+ */
+NGHTTP2_EXTERN void nghttp2_session_callbacks_set_send_data_callback(
+    nghttp2_session_callbacks *cbs,
+    nghttp2_send_data_callback send_data_callback);
+
 /**
  * @functypedef
  *

lib/nghttp2_callbacks.c

@@ -121,3 +121,9 @@ void nghttp2_session_callbacks_set_on_begin_frame_callback(
     nghttp2_on_begin_frame_callback on_begin_frame_callback) {
   cbs->on_begin_frame_callback = on_begin_frame_callback;
 }
+
+void nghttp2_session_callbacks_set_send_data_callback(
+    nghttp2_session_callbacks *cbs,
+    nghttp2_send_data_callback send_data_callback) {
+  cbs->send_data_callback = send_data_callback;
+}

lib/nghttp2_callbacks.h

@@ -106,6 +106,7 @@ struct nghttp2_session_callbacks {
    * Sets callback function invoked when a frame header is received.
    */
   nghttp2_on_begin_frame_callback on_begin_frame_callback;
+  nghttp2_send_data_callback send_data_callback;
 };
 
 #endif /* NGHTTP2_CALLBACKS_H */

lib/nghttp2_frame.c

@@ -813,7 +813,7 @@ int nghttp2_iv_check(const nghttp2_settings_entry *iv, size_t niv) {
   return 1;
 }
 
-static void frame_set_pad(nghttp2_buf *buf, size_t padlen) {
+static void frame_set_pad(nghttp2_buf *buf, size_t padlen, int framehd_only) {
   size_t trail_padlen;
   size_t newlen;
 
@@ -828,6 +828,10 @@ static void frame_set_pad(nghttp2_buf *buf, size_t padlen) {
   newlen = (nghttp2_get_uint32(buf->pos) >> 8) + padlen;
   nghttp2_put_uint32be(buf->pos, (uint32_t)((newlen << 8) + buf->pos[3]));
 
+  if (framehd_only) {
+    return;
+  }
+
   trail_padlen = padlen - 1;
   buf->pos[NGHTTP2_FRAME_HDLEN] = trail_padlen;
 
@@ -836,12 +840,10 @@ static void frame_set_pad(nghttp2_buf *buf, size_t padlen) {
   /* extend buffers trail_padlen bytes, since we ate previous padlen -
      trail_padlen byte(s) */
   buf->last += trail_padlen;
-
-  return;
 }
 
 int nghttp2_frame_add_pad(nghttp2_bufs *bufs, nghttp2_frame_hd *hd,
-                          size_t padlen) {
+                          size_t padlen, int framehd_only) {
   nghttp2_buf *buf;
 
   if (padlen == 0) {
@@ -874,7 +876,7 @@ int nghttp2_frame_add_pad(nghttp2_bufs *bufs, nghttp2_frame_hd *hd,
 
   assert(nghttp2_buf_avail(buf) >= (ssize_t)padlen - 1);
 
-  frame_set_pad(buf, padlen);
+  frame_set_pad(buf, padlen, framehd_only);
 
   hd->length += padlen;
   hd->flags |= NGHTTP2_FLAG_PADDED;

lib/nghttp2_frame.h

@@ -510,7 +510,8 @@ int nghttp2_iv_check(const nghttp2_settings_entry *iv, size_t niv);
  * Sets Pad Length field and flags and adjusts frame header position
  * of each buffers in |bufs|.  The number of padding is given in the
  * |padlen| including Pad Length field.  The |hd| is the frame header
- * for the serialized data.
+ * for the serialized data.  This function fills zeros padding region
+ * unless framehd_only is nonzero.
  *
  * This function returns 0 if it succeeds, or one of the following
  * negative error codes:
@@ -521,6 +522,6 @@ int nghttp2_iv_check(const nghttp2_settings_entry *iv, size_t niv);
  *     The length of the resulting frame is too large.
  */
 int nghttp2_frame_add_pad(nghttp2_bufs *bufs, nghttp2_frame_hd *hd,
-                          size_t padlen);
+                          size_t padlen, int framehd_only);
 
 #endif /* NGHTTP2_FRAME_H */

lib/nghttp2_outbound_item.h

@@ -74,6 +74,10 @@ typedef struct {
    * |eof| is 0. It becomes 1 after all data were read.
    */
   uint8_t eof;
+  /**
+   * The flag to indicate that NGHTTP2_DATA_FLAG_NO_COPY is used.
+   */
+  uint8_t no_copy;
 } nghttp2_data_aux_data;
 
 typedef enum {

lib/nghttp2_session.c

@@ -1701,7 +1701,7 @@ static int session_headers_add_pad(nghttp2_session *session,
   DEBUGF(fprintf(stderr, "send: padding selected: payloadlen=%zd, padlen=%zu\n",
                  padded_payloadlen, padlen));
 
-  rv = nghttp2_frame_add_pad(framebufs, &frame->hd, padlen);
+  rv = nghttp2_frame_add_pad(framebufs, &frame->hd, padlen, 0);
 
   if (rv != 0) {
     return rv;
@@ -2558,6 +2558,9 @@ static int session_after_frame_sent2(nghttp2_session *session) {
       return 0;
     }
 
+    /* Reset no_copy here because next write may not use this. */
+    aux_data->no_copy = 0;
+
     stream = nghttp2_session_get_stream(session, frame->hd.stream_id);
 
     /* If session is closed or RST_STREAM was queued, we won't send
@@ -2648,6 +2651,7 @@ static int session_after_frame_sent2(nghttp2_session *session) {
 
         return 0;
       }
+
       if (rv == NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE) {
         /* Stop DATA frame chain and issue RST_STREAM to close the
            stream.  We don't return
@@ -2671,6 +2675,12 @@ static int session_after_frame_sent2(nghttp2_session *session) {
       }
       assert(rv == 0);
 
+      if (aux_data->no_copy) {
+        aob->state = NGHTTP2_OB_SEND_NO_COPY;
+      } else {
+        aob->state = NGHTTP2_OB_SEND_DATA;
+      }
+
       return 0;
     }
 
@@ -2693,6 +2703,32 @@ static int session_after_frame_sent2(nghttp2_session *session) {
   return 0;
 }
 
+static int session_call_send_data(nghttp2_session *session,
+                                  nghttp2_outbound_item *item,
+                                  nghttp2_bufs *framebufs) {
+  int rv;
+  nghttp2_buf *buf;
+  size_t length;
+  nghttp2_frame *frame;
+  nghttp2_data_aux_data *aux_data;
+
+  buf = &framebufs->cur->buf;
+  frame = &item->frame;
+  length = frame->hd.length - frame->data.padlen;
+  aux_data = &item->aux_data.data;
+
+  rv = session->callbacks.send_data_callback(session, frame, buf->pos, length,
+                                             &aux_data->data_prd.source,
+                                             session->user_data);
+
+  if (rv == 0 || rv == NGHTTP2_ERR_WOULDBLOCK ||
+      rv == NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE) {
+    return rv;
+  }
+
+  return NGHTTP2_ERR_CALLBACK_FAILURE;
+}
+
 static ssize_t nghttp2_session_mem_send_internal(nghttp2_session *session,
                                                  const uint8_t **data_ptr,
                                                  int fast_cb) {
@@ -2823,6 +2859,11 @@ static ssize_t nghttp2_session_mem_send_internal(nghttp2_session *session,
         }
       } else {
         DEBUGF(fprintf(stderr, "send: next frame: DATA\n"));
+
+        if (item->aux_data.data.no_copy) {
+          aob->state = NGHTTP2_OB_SEND_NO_COPY;
+          break;
+        }
       }
 
       DEBUGF(fprintf(stderr,
@@ -2873,6 +2914,60 @@ static ssize_t nghttp2_session_mem_send_internal(nghttp2_session *session,
 
       return datalen;
     }
+    case NGHTTP2_OB_SEND_NO_COPY:
+      DEBUGF(fprintf(stderr, "send: no copy DATA\n"));
+
+      rv = session_call_send_data(session, aob->item, framebufs);
+      if (nghttp2_is_fatal(rv)) {
+        return rv;
+      }
+
+      if (rv == NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE) {
+        nghttp2_stream *stream;
+        nghttp2_frame *frame;
+
+        frame = &aob->item->frame;
+
+        stream = nghttp2_session_get_stream(session, frame->hd.stream_id);
+        if (stream) {
+          rv = nghttp2_stream_detach_item(stream, session);
+
+          if (nghttp2_is_fatal(rv)) {
+            return rv;
+          }
+
+          rv = nghttp2_session_add_rst_stream(session, frame->hd.stream_id,
+                                              NGHTTP2_INTERNAL_ERROR);
+          if (nghttp2_is_fatal(rv)) {
+            return rv;
+          }
+        }
+
+        active_outbound_item_reset(aob, mem);
+
+        break;
+      }
+
+      if (rv == NGHTTP2_ERR_WOULDBLOCK) {
+        return 0;
+      }
+
+      assert(rv == 0);
+
+      rv = session_after_frame_sent1(session);
+      if (rv < 0) {
+        assert(nghttp2_is_fatal(rv));
+        return rv;
+      }
+      rv = session_after_frame_sent2(session);
+      if (rv < 0) {
+        assert(nghttp2_is_fatal(rv));
+        return rv;
+      }
+
+      /* We have already adjusted the next state */
+
+      break;
     }
   }
 }
@@ -6223,6 +6318,17 @@ int nghttp2_session_pack_data(nghttp2_session *session, nghttp2_bufs *bufs,
     }
   }
 
+  if (data_flags & NGHTTP2_DATA_FLAG_NO_COPY) {
+    if (session->callbacks.send_data_callback == NULL) {
+      DEBUGF(fprintf(
+          stderr,
+          "NGHTTP2_DATA_FLAG_NO_COPY requires send_data_callback set\n"));
+
+      return NGHTTP2_ERR_CALLBACK_FAILURE;
+    }
+    aux_data->no_copy = 1;
+  }
+
   frame->hd.length = payloadlen;
   frame->data.padlen = 0;
 
@@ -6239,7 +6345,8 @@ int nghttp2_session_pack_data(nghttp2_session *session, nghttp2_bufs *bufs,
 
   nghttp2_frame_pack_frame_hd(buf->pos, &frame->hd);
 
-  rv = nghttp2_frame_add_pad(bufs, &frame->hd, frame->data.padlen);
+  rv = nghttp2_frame_add_pad(bufs, &frame->hd, frame->data.padlen,
+                             aux_data->no_copy);
   if (rv != 0) {
     return rv;
   }

lib/nghttp2_session.h

@@ -52,7 +52,8 @@ typedef enum {
 
 typedef enum {
   NGHTTP2_OB_POP_ITEM,
-  NGHTTP2_OB_SEND_DATA
+  NGHTTP2_OB_SEND_DATA,
+  NGHTTP2_OB_SEND_NO_COPY
 } nghttp2_outbound_state;
 
 typedef struct {

src/HttpServer.cc

@@ -371,6 +371,8 @@ struct ev_loop *Http2Handler::get_loop() const {
   return sessions_->get_loop();
 }
 
+Http2Handler::WriteBuf *Http2Handler::get_wb() { return &wb_; }
+
 int Http2Handler::setup_bev() { return 0; }
 
 int Http2Handler::fill_wb() {
@@ -833,18 +835,9 @@ ssize_t file_read_callback(nghttp2_session *session, int32_t stream_id,
   auto hd = static_cast<Http2Handler *>(user_data);
   auto stream = hd->get_stream(stream_id);
 
-  int fd = source->fd;
+  size_t nread = std::min(stream->body_left, static_cast<int64_t>(length));
-  ssize_t nread;
 
-  while ((nread = read(fd, buf, length)) == -1 && errno == EINTR)
+  *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY;
-    ;
-
-  if (nread == -1) {
-    remove_stream_read_timeout(stream);
-    remove_stream_write_timeout(stream);
-
-    return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
-  }
 
   stream->body_left -= nread;
   if (nread == 0 || stream->body_left <= 0) {
@@ -1220,6 +1213,62 @@ int hd_on_frame_send_callback(nghttp2_session *session,
 }
 } // namespace
 
+namespace {
+int send_data_callback(nghttp2_session *session, nghttp2_frame *frame,
+                       const uint8_t *framehd, size_t length,
+                       nghttp2_data_source *source, void *user_data) {
+  auto hd = static_cast<Http2Handler *>(user_data);
+  auto stream = hd->get_stream(frame->hd.stream_id);
+
+  if (!stream) {
+    return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+  }
+
+  auto wb = hd->get_wb();
+  auto padlen = frame->data.padlen;
+
+  if (wb->wleft() < 9 + length + padlen) {
+    return NGHTTP2_ERR_WOULDBLOCK;
+  }
+
+  int fd = source->fd;
+
+  auto p = wb->last;
+
+  p = std::copy_n(framehd, 9, p);
+
+  if (padlen) {
+    *p++ = padlen - 1;
+  }
+
+  while (length) {
+    ssize_t nread;
+    while ((nread = read(fd, p, length)) == -1 && errno == EINTR)
+      ;
+
+    if (nread == -1) {
+      auto stream = hd->get_stream(frame->hd.stream_id);
+      remove_stream_read_timeout(stream);
+      remove_stream_write_timeout(stream);
+
+      return NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
+    }
+
+    length -= nread;
+    p += nread;
+  }
+
+  if (padlen) {
+    std::fill(p, p + padlen - 1, 0);
+    p += padlen - 1;
+  }
+
+  wb->last = p;
+
+  return 0;
+}
+} // namespace
+
 namespace {
 ssize_t select_padding_callback(nghttp2_session *session,
                                 const nghttp2_frame *frame, size_t max_payload,
@@ -1288,6 +1337,9 @@ void fill_callback(nghttp2_session_callbacks *callbacks, const Config *config) {
   nghttp2_session_callbacks_set_on_begin_headers_callback(
       callbacks, on_begin_headers_callback);
 
+  nghttp2_session_callbacks_set_send_data_callback(callbacks,
+                                                   send_data_callback);
+
   if (config->padding) {
     nghttp2_session_callbacks_set_select_padding_callback(
         callbacks, select_padding_callback);

src/HttpServer.h

@@ -140,12 +140,16 @@ public:
 
   struct ev_loop *get_loop() const;
 
+  using WriteBuf = Buffer<65536>;
+
+  WriteBuf *get_wb();
+
 private:
   ev_io wev_;
   ev_io rev_;
   ev_timer settings_timerev_;
   std::map<int32_t, std::unique_ptr<Stream>> id2stream_;
-  Buffer<65536> wb_;
+  WriteBuf wb_;
   std::function<int(Http2Handler &)> read_, write_;
   int64_t session_id_;
   nghttp2_session *session_;

tests/main.c

@@ -260,6 +260,8 @@ int main(int argc _U_, char *argv[] _U_) {
                    test_nghttp2_session_cancel_reserved_remote) ||
       !CU_add_test(pSuite, "session_reset_pending_headers",
                    test_nghttp2_session_reset_pending_headers) ||
+      !CU_add_test(pSuite, "session_send_data_callback",
+                   test_nghttp2_session_send_data_callback) ||
       !CU_add_test(pSuite, "http_mandatory_headers",
                    test_nghttp2_http_mandatory_headers) ||
       !CU_add_test(pSuite, "http_content_length",

tests/nghttp2_session_test.c

@@ -294,6 +294,50 @@ static ssize_t no_end_stream_data_source_read_callback(
   return 0;
 }
 
+static ssize_t no_copy_data_source_read_callback(
+    nghttp2_session *session _U_, int32_t stream_id _U_, uint8_t *buf _U_,
+    size_t len, uint32_t *data_flags, nghttp2_data_source *source _U_,
+    void *user_data) {
+  my_user_data *ud = (my_user_data *)user_data;
+  size_t wlen;
+  if (len < ud->data_source_length) {
+    wlen = len;
+  } else {
+    wlen = ud->data_source_length;
+  }
+
+  ud->data_source_length -= wlen;
+
+  *data_flags |= NGHTTP2_DATA_FLAG_NO_COPY;
+
+  if (ud->data_source_length == 0) {
+    *data_flags |= NGHTTP2_DATA_FLAG_EOF;
+  }
+  return wlen;
+}
+
+static int send_data_callback(nghttp2_session *session _U_,
+                              nghttp2_frame *frame, const uint8_t *framehd,
+                              size_t length, nghttp2_data_source *source _U_,
+                              void *user_data) {
+  accumulator *acc = ((my_user_data *)user_data)->acc;
+
+  memcpy(acc->buf + acc->length, framehd, NGHTTP2_FRAME_HDLEN);
+  acc->length += NGHTTP2_FRAME_HDLEN;
+
+  if (frame->data.padlen) {
+    *(acc->buf + acc->length++) = frame->data.padlen - 1;
+  }
+
+  acc->length += length;
+
+  if (frame->data.padlen) {
+    acc->length += frame->data.padlen - 1;
+  }
+
+  return 0;
+}
+
 /* static void no_stream_user_data_stream_close_callback */
 /* (nghttp2_session *session, */
 /*  int32_t stream_id, */
@@ -2337,8 +2381,7 @@ void test_nghttp2_session_on_push_promise_received(void) {
   nghttp2_session_client_new(&session, &callbacks, &user_data);
 
   nghttp2_session_open_stream(session, 2, NGHTTP2_STREAM_FLAG_NONE,
-                              &pri_spec_default,
+                              &pri_spec_default, NGHTTP2_STREAM_RESERVED, NULL);
-                              NGHTTP2_STREAM_RESERVED, NULL);
   /* Attempt to PUSH_PROMISE against reserved (remote) stream */
   nghttp2_frame_push_promise_init(&frame.push_promise, NGHTTP2_FLAG_END_HEADERS,
                                   2, 4, NULL, 0);
@@ -2358,8 +2401,7 @@ void test_nghttp2_session_on_push_promise_received(void) {
   nghttp2_session_client_new(&session, &callbacks, &user_data);
 
   nghttp2_session_open_stream(session, 1, NGHTTP2_STREAM_FLAG_NONE,
-                              &pri_spec_default,
+                              &pri_spec_default, NGHTTP2_STREAM_OPENING, NULL);
-                              NGHTTP2_STREAM_OPENING, NULL);
 
   session->local_settings.enable_push = 0;
 
@@ -2381,8 +2423,7 @@ void test_nghttp2_session_on_push_promise_received(void) {
   nghttp2_session_client_new(&session, &callbacks, &user_data);
 
   nghttp2_session_open_stream(session, 1, NGHTTP2_STREAM_FLAG_NONE,
-                              &pri_spec_default,
+                              &pri_spec_default, NGHTTP2_STREAM_OPENING, NULL);
-                              NGHTTP2_STREAM_OPENING, NULL);
   nvlen = ARRLEN(malformed_nva);
   nghttp2_nv_array_copy(&nva, malformed_nva, nvlen, mem);
   nghttp2_frame_push_promise_init(&frame.push_promise, NGHTTP2_FLAG_END_HEADERS,
@@ -2402,8 +2443,7 @@ void test_nghttp2_session_on_push_promise_received(void) {
   nghttp2_session_client_new(&session, &callbacks, &user_data);
 
   nghttp2_session_open_stream(session, 1, NGHTTP2_STREAM_FLAG_NONE,
-                              &pri_spec_default,
+                              &pri_spec_default, NGHTTP2_STREAM_OPENING, NULL);
-                              NGHTTP2_STREAM_OPENING, NULL);
   /* Submit settings with ENABLE_PUSH = 0 (thus disabling push) */
   nghttp2_submit_settings(session, NGHTTP2_FLAG_NONE, &iv, 1);
 
@@ -2552,8 +2592,7 @@ void test_nghttp2_session_on_window_update_received(void) {
   /* Receiving WINDOW_UPDATE on reserved (remote) stream is a
      connection error */
   nghttp2_session_open_stream(session, 2, NGHTTP2_STREAM_FLAG_NONE,
-                              &pri_spec_default,
+                              &pri_spec_default, NGHTTP2_STREAM_RESERVED, NULL);
-                              NGHTTP2_STREAM_RESERVED, NULL);
 
   nghttp2_frame_window_update_init(&frame.window_update, NGHTTP2_FLAG_NONE, 2,
                                    4096);
@@ -2613,8 +2652,7 @@ void test_nghttp2_session_on_data_received(void) {
 
   /* If NGHTTP2_STREAM_CLOSING state, DATA frame is discarded. */
   nghttp2_session_open_stream(session, 4, NGHTTP2_STREAM_FLAG_NONE,
-                              &pri_spec_default,
+                              &pri_spec_default, NGHTTP2_STREAM_CLOSING, NULL);
-                              NGHTTP2_STREAM_CLOSING, NULL);
 
   frame.hd.flags = NGHTTP2_FLAG_NONE;
   frame.hd.stream_id = 4;
@@ -3644,8 +3682,7 @@ void test_nghttp2_submit_headers_push_reply(void) {
      error */
   CU_ASSERT(0 == nghttp2_session_client_new(&session, &callbacks, &ud));
   nghttp2_session_open_stream(session, 2, NGHTTP2_STREAM_FLAG_NONE,
-                              &pri_spec_default,
+                              &pri_spec_default, NGHTTP2_STREAM_RESERVED, NULL);
-                              NGHTTP2_STREAM_RESERVED, NULL);
   CU_ASSERT(0 == nghttp2_submit_headers(session, NGHTTP2_FLAG_NONE, 2, NULL,
                                         reqnv, ARRLEN(reqnv), NULL));
 
@@ -3911,8 +3948,7 @@ void test_nghttp2_submit_settings_update_local_window_size(void) {
   stream->recv_window_size = 32768;
 
   nghttp2_session_open_stream(session, 3, NGHTTP2_STREAM_FLAG_NONE,
-                              &pri_spec_default, NGHTTP2_STREAM_OPENED,
+                              &pri_spec_default, NGHTTP2_STREAM_OPENED, NULL);
-                              NULL);
 
   CU_ASSERT(0 == nghttp2_submit_settings(session, NGHTTP2_FLAG_NONE, iv, 1));
   CU_ASSERT(0 == nghttp2_session_send(session));
@@ -6950,6 +6986,51 @@ void test_nghttp2_session_reset_pending_headers(void) {
   nghttp2_session_del(session);
 }
 
+void test_nghttp2_session_send_data_callback(void) {
+  nghttp2_session *session;
+  nghttp2_session_callbacks callbacks;
+  nghttp2_data_provider data_prd;
+  my_user_data ud;
+  accumulator acc;
+  nghttp2_frame_hd hd;
+
+  memset(&callbacks, 0, sizeof(nghttp2_session_callbacks));
+  callbacks.send_callback = accumulator_send_callback;
+  callbacks.send_data_callback = send_data_callback;
+
+  data_prd.read_callback = no_copy_data_source_read_callback;
+
+  acc.length = 0;
+  ud.acc = &acc;
+
+  ud.data_source_length = NGHTTP2_DATA_PAYLOADLEN * 2;
+
+  nghttp2_session_client_new(&session, &callbacks, &ud);
+
+  nghttp2_session_open_stream(session, 1, NGHTTP2_STREAM_FLAG_NONE,
+                              &pri_spec_default, NGHTTP2_STREAM_OPENING, NULL);
+
+  nghttp2_submit_data(session, NGHTTP2_FLAG_END_STREAM, 1, &data_prd);
+
+  CU_ASSERT(0 == nghttp2_session_send(session));
+
+  CU_ASSERT((NGHTTP2_FRAME_HDLEN + NGHTTP2_DATA_PAYLOADLEN) * 2 == acc.length);
+
+  nghttp2_frame_unpack_frame_hd(&hd, acc.buf);
+
+  CU_ASSERT(16384 == hd.length);
+  CU_ASSERT(NGHTTP2_DATA == hd.type);
+  CU_ASSERT(NGHTTP2_FLAG_NONE == hd.flags);
+
+  nghttp2_frame_unpack_frame_hd(&hd, acc.buf + NGHTTP2_FRAME_HDLEN + hd.length);
+
+  CU_ASSERT(16384 == hd.length);
+  CU_ASSERT(NGHTTP2_DATA == hd.type);
+  CU_ASSERT(NGHTTP2_FLAG_END_STREAM == hd.flags);
+
+  nghttp2_session_del(session);
+}
+
 static void check_nghttp2_http_recv_headers_fail(
     nghttp2_session *session, nghttp2_hd_deflater *deflater, int32_t stream_id,
     int stream_state, const nghttp2_nv *nva, size_t nvlen) {

tests/nghttp2_session_test.h

@@ -123,6 +123,7 @@ void test_nghttp2_session_delete_data_item(void);
 void test_nghttp2_session_open_idle_stream(void);
 void test_nghttp2_session_cancel_reserved_remote(void);
 void test_nghttp2_session_reset_pending_headers(void);
+void test_nghttp2_session_send_data_callback(void);
 void test_nghttp2_http_mandatory_headers(void);
 void test_nghttp2_http_content_length(void);
 void test_nghttp2_http_content_length_mismatch(void);