granicus.if.org Git - apache/blob - modules/http2/h2_mplx.h

   1 /* Copyright 2015 greenbytes GmbH (https://www.greenbytes.de)
   2  *
   3  * Licensed under the Apache License, Version 2.0 (the "License");
   4  * you may not use this file except in compliance with the License.
   5  * You may obtain a copy of the License at
   6  *
   7  * http://www.apache.org/licenses/LICENSE-2.0
   8
   9  * Unless required by applicable law or agreed to in writing, software
  10  * distributed under the License is distributed on an "AS IS" BASIS,
  11  * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12  * See the License for the specific language governing permissions and
  13  * limitations under the License.
  14  */
  15
  16 #ifndef __mod_h2__h2_mplx__
  17 #define __mod_h2__h2_mplx__
  18
  19 /**
  20  * The stream multiplexer. It pushes buckets from the connection
  21  * thread to the stream task threads and vice versa. It's thread-safe
  22  * to use.
  23  *
  24  * There is one h2_mplx instance for each h2_session, which sits on top
  25  * of a particular httpd conn_rec. Input goes from the connection to
  26  * the stream tasks. Output goes from the stream tasks to the connection,
  27  * e.g. the client.
  28  *
  29  * For each stream, there can be at most "H2StreamMaxMemSize" output bytes
  30  * queued in the multiplexer. If a task thread tries to write more
  31  * data, it is blocked until space becomes available.
  32  *
  33  * Writing input is never blocked. In order to use flow control on the input,
  34  * the mplx can be polled for input data consumption.
  35  */
  36
  37 struct apr_pool_t;
  38 struct apr_thread_mutex_t;
  39 struct apr_thread_cond_t;
  40 struct h2_config;
  41 struct h2_response;
  42 struct h2_task;
  43 struct h2_stream;
  44 struct h2_io_set;
  45 struct apr_thread_cond_t;
  46 struct h2_workers;
  47 struct h2_stream_set;
  48 struct h2_task_queue;
  49
  50 #include "h2_io.h"
  51
  52 typedef struct h2_mplx h2_mplx;
  53
  54 struct h2_mplx {
  55     long id;
  56     APR_RING_ENTRY(h2_mplx) link;
  57     volatile apr_uint32_t refs;
  58     conn_rec *c;
  59     apr_pool_t *pool;
  60     apr_bucket_alloc_t *bucket_alloc;
  61
  62     struct h2_task_queue *q;
  63     struct h2_io_set *stream_ios;
  64     struct h2_io_set *ready_ios;
  65
  66     apr_thread_mutex_t *lock;
  67     struct apr_thread_cond_t *added_output;
  68     struct apr_thread_cond_t *join_wait;
  69
  70     int aborted;
  71     apr_size_t stream_max_mem;
  72
  73     apr_pool_t *spare_pool;           /* spare pool, ready for next stream */
  74     struct h2_stream_set *closed;     /* streams closed, but task ongoing */
  75     struct h2_workers *workers;
  76     int file_handles_allowed;
  77 };
  78
  79 /*******************************************************************************
  80  * Object lifecycle and information.
  81  ******************************************************************************/
  82
  83 /**
  84  * Create the multiplexer for the given HTTP2 session.
  85  * Implicitly has reference count 1.
  86  */
  87 h2_mplx *h2_mplx_create(conn_rec *c, apr_pool_t *master,
  88                         struct h2_workers *workers);
  89
  90 /**
  91  * Increase the reference counter of this mplx.
  92  */
  93 void h2_mplx_reference(h2_mplx *m);
  94
  95 /**
  96  * Decreases the reference counter of this mplx.
  97  */
  98 void h2_mplx_release(h2_mplx *m);
  99 /**
 100  * Decreases the reference counter of this mplx and waits for it
 101  * to reached 0, destroy the mplx afterwards.
 102  * This is to be called from the thread that created the mplx in
 103  * the first place.
 104  * @param m the mplx to be released and destroyed
 105  * @param wait condition var to wait on for ref counter == 0
 106  */
 107 apr_status_t h2_mplx_release_and_join(h2_mplx *m, struct apr_thread_cond_t *wait);
 108
 109 /**
 110  * Aborts the multiplexer. It will answer all future invocation with
 111  * APR_ECONNABORTED, leading to early termination of ongoing tasks.
 112  */
 113 void h2_mplx_abort(h2_mplx *mplx);
 114
 115 void h2_mplx_task_done(h2_mplx *m, int stream_id);
 116
 117 /*******************************************************************************
 118  * IO lifetime of streams.
 119  ******************************************************************************/
 120 /**
 121  * Prepares the multiplexer to handle in-/output on the given stream id.
 122  */
 123 struct h2_stream *h2_mplx_open_io(h2_mplx *mplx, int stream_id);
 124
 125 /**
 126  * Ends cleanup of a stream in sync with execution thread.
 127  */
 128 apr_status_t h2_mplx_cleanup_stream(h2_mplx *m, struct h2_stream *stream);
 129
 130 /* Return != 0 iff the multiplexer has data for the given stream.
 131  */
 132 int h2_mplx_out_has_data_for(h2_mplx *m, int stream_id);
 133
 134 /**
 135  * Waits on output data from any stream in this session to become available.
 136  * Returns APR_TIMEUP if no data arrived in the given time.
 137  */
 138 apr_status_t h2_mplx_out_trywait(h2_mplx *m, apr_interval_time_t timeout,
 139                                  struct apr_thread_cond_t *iowait);
 140
 141 /*******************************************************************************
 142  * Stream processing.
 143  ******************************************************************************/
 144
 145 /**
 146  * Perform the task on the given stream.
 147  */
 148 apr_status_t h2_mplx_do_task(h2_mplx *mplx, struct h2_task *task);
 149
 150 struct h2_task *h2_mplx_pop_task(h2_mplx *mplx, int *has_more);
 151
 152 apr_status_t h2_mplx_create_task(h2_mplx *mplx, struct h2_stream *stream);
 153
 154 /*******************************************************************************
 155  * Input handling of streams.
 156  ******************************************************************************/
 157
 158 /**
 159  * Reads a buckets for the given stream_id. Will return ARP_EAGAIN when
 160  * called with APR_NONBLOCK_READ and no data present. Will return APR_EOF
 161  * when the end of the stream input has been reached.
 162  * The condition passed in will be used for blocking/signalling and will
 163  * be protected by the mplx's own mutex.
 164  */
 165 apr_status_t h2_mplx_in_read(h2_mplx *m, apr_read_type_e block,
 166                              int stream_id, apr_bucket_brigade *bb,
 167                              struct apr_thread_cond_t *iowait);
 168
 169 /**
 170  * Appends data to the input of the given stream. Storage of input data is
 171  * not subject to flow control.
 172  */
 173 apr_status_t h2_mplx_in_write(h2_mplx *mplx, int stream_id,
 174                               apr_bucket_brigade *bb);
 175
 176 /**
 177  * Closes the input for the given stream_id.
 178  */
 179 apr_status_t h2_mplx_in_close(h2_mplx *m, int stream_id);
 180
 181 /**
 182  * Returns != 0 iff the input for the given stream has been closed. There
 183  * could still be data queued, but it can be read without blocking.
 184  */
 185 int h2_mplx_in_has_eos_for(h2_mplx *m, int stream_id);
 186
 187 /**
 188  * Callback invoked for every stream that had input data read since
 189  * the last invocation.
 190  */
 191 typedef void h2_mplx_consumed_cb(void *ctx, int stream_id, apr_size_t consumed);
 192
 193 /**
 194  * Invoke the callback for all streams that had bytes read since the last
 195  * call to this function. If no stream had input data consumed, the callback
 196  * is not invoked.
 197  * Returns APR_SUCCESS when an update happened, APR_EAGAIN if no update
 198  * happened.
 199  */
 200 apr_status_t h2_mplx_in_update_windows(h2_mplx *m,
 201                                        h2_mplx_consumed_cb *cb, void *ctx);
 202
 203 /*******************************************************************************
 204  * Output handling of streams.
 205  ******************************************************************************/
 206
 207 /**
 208  * Get a stream whose response is ready for submit. Will set response and
 209  * any out data available in stream.
 210  * @param m the mplxer to get a response from
 211  * @param bb the brigade to place any existing repsonse body data into
 212  */
 213 struct h2_stream *h2_mplx_next_submit(h2_mplx *m,
 214                                       struct h2_stream_set *streams);
 215
 216 /**
 217  * Reads output data from the given stream. Will never block, but
 218  * return APR_EAGAIN until data arrives or the stream is closed.
 219  */
 220 apr_status_t h2_mplx_out_readx(h2_mplx *mplx, int stream_id,
 221                                h2_io_data_cb *cb, void *ctx,
 222                                apr_size_t *plen, int *peos);
 223
 224 /**
 225  * Opens the output for the given stream with the specified response.
 226  */
 227 apr_status_t h2_mplx_out_open(h2_mplx *mplx, int stream_id,
 228                               struct h2_response *response,
 229                               ap_filter_t* filter, apr_bucket_brigade *bb,
 230                               struct apr_thread_cond_t *iowait);
 231
 232 /**
 233  * Append the brigade to the stream output. Might block if amount
 234  * of bytes buffered reaches configured max.
 235  * @param stream_id the stream identifier
 236  * @param filter the apache filter context of the data
 237  * @param bb the bucket brigade to append
 238  * @param iowait a conditional used for block/signalling in h2_mplx
 239  */
 240 apr_status_t h2_mplx_out_write(h2_mplx *mplx, int stream_id,
 241                                ap_filter_t* filter, apr_bucket_brigade *bb,
 242                                struct apr_thread_cond_t *iowait);
 243
 244 /**
 245  * Closes the output stream. Readers of this stream will get all pending
 246  * data and then only APR_EOF as result.
 247  */
 248 apr_status_t h2_mplx_out_close(h2_mplx *m, int stream_id);
 249
 250 /*******************************************************************************
 251  * h2_mplx list Manipulation.
 252  ******************************************************************************/
 253
 254 /**
 255  * The magic pointer value that indicates the head of a h2_mplx list
 256  * @param  b The mplx list
 257  * @return The magic pointer value
 258  */
 259 #define H2_MPLX_LIST_SENTINEL(b)        APR_RING_SENTINEL((b), h2_mplx, link)
 260
 261 /**
 262  * Determine if the mplx list is empty
 263  * @param b The list to check
 264  * @return true or false
 265  */
 266 #define H2_MPLX_LIST_EMPTY(b)   APR_RING_EMPTY((b), h2_mplx, link)
 267
 268 /**
 269  * Return the first mplx in a list
 270  * @param b The list to query
 271  * @return The first mplx in the list
 272  */
 273 #define H2_MPLX_LIST_FIRST(b)   APR_RING_FIRST(b)
 274
 275 /**
 276  * Return the last mplx in a list
 277  * @param b The list to query
 278  * @return The last mplx int he list
 279  */
 280 #define H2_MPLX_LIST_LAST(b)    APR_RING_LAST(b)
 281
 282 /**
 283  * Insert a single mplx at the front of a list
 284  * @param b The list to add to
 285  * @param e The mplx to insert
 286  */
 287 #define H2_MPLX_LIST_INSERT_HEAD(b, e) do {                             \
 288 h2_mplx *ap__b = (e);                                        \
 289 APR_RING_INSERT_HEAD((b), ap__b, h2_mplx, link);        \
 290 } while (0)
 291
 292 /**
 293  * Insert a single mplx at the end of a list
 294  * @param b The list to add to
 295  * @param e The mplx to insert
 296  */
 297 #define H2_MPLX_LIST_INSERT_TAIL(b, e) do {                             \
 298 h2_mplx *ap__b = (e);                                   \
 299 APR_RING_INSERT_TAIL((b), ap__b, h2_mplx, link);        \
 300 } while (0)
 301
 302 /**
 303  * Get the next mplx in the list
 304  * @param e The current mplx
 305  * @return The next mplx
 306  */
 307 #define H2_MPLX_NEXT(e) APR_RING_NEXT((e), link)
 308 /**
 309  * Get the previous mplx in the list
 310  * @param e The current mplx
 311  * @return The previous mplx
 312  */
 313 #define H2_MPLX_PREV(e) APR_RING_PREV((e), link)
 314
 315 /**
 316  * Remove a mplx from its list
 317  * @param e The mplx to remove
 318  */
 319 #define H2_MPLX_REMOVE(e)       APR_RING_REMOVE((e), link)
 320
 321
 322 #endif /* defined(__mod_h2__h2_mplx__) */