/* Licensed to the Apache Software Foundation (ASF) under one or more
* contributor license agreements. See the NOTICE file distributed with
* this work for additional information regarding copyright ownership.
* The ASF licenses this file to You under the Apache License, Version 2.0
* (the "License"); you may not use this file except in compliance with
* the License. You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/
#ifndef h2_bucket_beam_h
#define h2_bucket_beam_h
struct apr_thread_mutex_t;
struct apr_thread_cond_t;
/*******************************************************************************
* apr_bucket list without bells and whistles
******************************************************************************/
/**
* h2_blist can hold a list of buckets just like apr_bucket_brigade, but
* does not to any allocations or related features.
*/
typedef struct {
APR_RING_HEAD(h2_bucket_list, apr_bucket) list;
} h2_blist;
#define H2_BLIST_INIT(b) APR_RING_INIT(&(b)->list, apr_bucket, link);
#define H2_BLIST_SENTINEL(b) APR_RING_SENTINEL(&(b)->list, apr_bucket, link)
#define H2_BLIST_EMPTY(b) APR_RING_EMPTY(&(b)->list, apr_bucket, link)
#define H2_BLIST_FIRST(b) APR_RING_FIRST(&(b)->list)
#define H2_BLIST_LAST(b) APR_RING_LAST(&(b)->list)
#define H2_BLIST_INSERT_HEAD(b, e) do { \
apr_bucket *ap__b = (e); \
APR_RING_INSERT_HEAD(&(b)->list, ap__b, apr_bucket, link); \
} while (0)
#define H2_BLIST_INSERT_TAIL(b, e) do { \
apr_bucket *ap__b = (e); \
APR_RING_INSERT_TAIL(&(b)->list, ap__b, apr_bucket, link); \
} while (0)
#define H2_BLIST_CONCAT(a, b) do { \
APR_RING_CONCAT(&(a)->list, &(b)->list, apr_bucket, link); \
} while (0)
#define H2_BLIST_PREPEND(a, b) do { \
APR_RING_PREPEND(&(a)->list, &(b)->list, apr_bucket, link); \
} while (0)
/*******************************************************************************
* h2_bucket_beam
******************************************************************************/
/**
* A h2_bucket_beam solves the task of transferring buckets, esp. their data,
* across threads with zero buffer copies.
*
* When a thread, let's call it the sender thread, wants to send buckets to
* another, the green thread, it creates a h2_bucket_beam and adds buckets
* via the h2_beam_send(). It gives the beam to the green thread which then
* can receive buckets into its own brigade via h2_beam_receive().
*
* Sending and receiving can happen concurrently.
*
* The beam can limit the amount of data it accepts via the buffer_size. This
* can also be adjusted during its lifetime. Sends and receives can be done blocking.
* A timeout can be set for such blocks.
*
* Care needs to be taken when terminating the beam. The beam registers at
* the pool it was created with and will cleanup after itself. However, if
* received buckets do still exist, already freed memory might be accessed.
* The beam does a assertion on this condition.
*
* The proper way of shutting down a beam is to first make sure there are no
* more green buckets out there, then cleanup the beam to purge eventually
* still existing sender buckets and then, possibly, terminate the beam itself
* (or the pool it was created with).
*
* The following restrictions apply to bucket transport:
* - only EOS and FLUSH meta buckets are copied through. All other meta buckets
* are kept in the beams hold.
* - all kind of data buckets are transported through:
* - transient buckets are converted to heap ones on send
* - heap and pool buckets require no extra handling
* - buckets with indeterminate length are read on send
* - file buckets will transfer the file itself into a new bucket, if allowed
* - all other buckets are read on send to make sure data is present
*
* This assures that when the sender thread sends its sender buckets, the data
* is made accessible while still on the sender side. The sender bucket then enters
* the beams hold storage.
* When the green thread calls receive, sender buckets in the hold are wrapped
* into special beam buckets. Beam buckets on read present the data directly
* from the internal sender one, but otherwise live on the green side. When a
* beam bucket gets destroyed, it notifies its beam that the corresponding
* sender bucket from the hold may be destroyed.
* Since the destruction of green buckets happens in the green thread, any
* corresponding sender bucket can not immediately be destroyed, as that would
* result in race conditions.
* Instead, the beam transfers such sender buckets from the hold to the purge
* storage. Next time there is a call from the sender side, the buckets in
* purge will be deleted.
*
* There are callbacks that can be registesender with a beam:
* - a "consumed" callback that gets called on the sender side with the
* amount of data that has been received by the green side. The amount
* is a delta from the last callback invocation. The sender side can trigger
* these callbacks by calling h2_beam_send() with a NULL brigade.
* - a "can_beam_file" callback that can prohibit the transfer of file handles
* through the beam. This will cause file buckets to be read on send and
* its data buffer will then be transports just like a heap bucket would.
* When no callback is registered, no restrictions apply and all files are
* passed through.
* File handles transfersender to the green side will stay there until the
* receiving brigade's pool is destroyed/cleared. If the pool lives very
* long or if many different files are beamed, the process might run out
* of available file handles.
*
* The name "beam" of course is inspired by good old transporter
* technology where humans are kept inside the transporter's memory
* buffers until the transmission is complete. Star gates use a similar trick.
*/
typedef void h2_beam_mutex_leave(void *ctx, struct apr_thread_mutex_t *lock);
typedef struct {
apr_thread_mutex_t *mutex;
h2_beam_mutex_leave *leave;
void *leave_ctx;
} h2_beam_lock;
typedef struct h2_bucket_beam h2_bucket_beam;
typedef apr_status_t h2_beam_mutex_enter(void *ctx, h2_beam_lock *pbl);
typedef void h2_beam_io_callback(void *ctx, h2_bucket_beam *beam,
apr_off_t bytes);
typedef void h2_beam_ev_callback(void *ctx, h2_bucket_beam *beam);
typedef struct h2_beam_proxy h2_beam_proxy;
typedef struct {
APR_RING_HEAD(h2_beam_proxy_list, h2_beam_proxy) list;
} h2_bproxy_list;
typedef int h2_beam_can_beam_callback(void *ctx, h2_bucket_beam *beam,
apr_file_t *file);
typedef enum {
H2_BEAM_OWNER_SEND,
H2_BEAM_OWNER_RECV
} h2_beam_owner_t;
/**
* Will deny all transfer of apr_file_t across the beam and force
* a data copy instead.
*/
int h2_beam_no_files(void *ctx, h2_bucket_beam *beam, apr_file_t *file);
struct h2_bucket_beam {
int id;
const char *tag;
apr_pool_t *pool;
h2_beam_owner_t owner;
h2_blist send_list;
h2_blist hold_list;
h2_blist purge_list;
apr_bucket_brigade *recv_buffer;
h2_bproxy_list proxies;
apr_pool_t *send_pool;
apr_pool_t *recv_pool;
apr_size_t max_buf_size;
apr_interval_time_t timeout;
apr_off_t sent_bytes; /* amount of bytes send */
apr_off_t received_bytes; /* amount of bytes received */
apr_size_t buckets_sent; /* # of beam buckets sent */
apr_size_t files_beamed; /* how many file handles have been set aside */
unsigned int aborted : 1;
unsigned int closed : 1;
unsigned int close_sent : 1;
unsigned int tx_mem_limits : 1; /* only memory size counts on transfers */
struct apr_thread_mutex_t *lock;
struct apr_thread_cond_t *change;
apr_off_t cons_bytes_reported; /* amount of bytes reported as consumed */
h2_beam_ev_callback *cons_ev_cb;
h2_beam_io_callback *cons_io_cb;
void *cons_ctx;
apr_off_t prod_bytes_reported; /* amount of bytes reported as produced */
h2_beam_io_callback *prod_io_cb;
void *prod_ctx;
h2_beam_can_beam_callback *can_beam_fn;
void *can_beam_ctx;
};
/**
* Creates a new bucket beam for transfer of buckets across threads.
*
* The pool the beam is created with will be protected by the given
* mutex and will be used in multiple threads. It needs a pool allocator
* that is only used inside that same mutex.
*
* @param pbeam will hold the created beam on return
* @param pool pool owning the beam, beam will cleanup when pool released
* @param id identifier of the beam
* @param tag tag identifying beam for logging
* @param owner if the beam is owned by the sender or receiver, e.g. if
* the pool owner is using this beam for sending or receiving
* @param buffer_size maximum memory footprint of buckets buffered in beam, or
* 0 for no limitation
* @param timeout timeout for blocking operations
*/
apr_status_t h2_beam_create(h2_bucket_beam **pbeam,
apr_pool_t *pool,
int id, const char *tag,
h2_beam_owner_t owner,
apr_size_t buffer_size,
apr_interval_time_t timeout);
/**
* Destroys the beam immediately without cleanup.
*/
apr_status_t h2_beam_destroy(h2_bucket_beam *beam);
/**
* Send buckets from the given brigade through the beam. Will hold buckets
* internally as long as they have not been processed by the receiving side.
* All accepted buckets are removed from the given brigade. Will return with
* APR_EAGAIN on non-blocking sends when not all buckets could be accepted.
*
* Call from the sender side only.
*/
apr_status_t h2_beam_send(h2_bucket_beam *beam,
apr_bucket_brigade *bb,
apr_read_type_e block);
/**
* Register the pool from which future buckets are send. This defines
* the lifetime of the buckets, e.g. the pool should not be cleared/destroyed
* until the data is no longer needed (or has been received).
*/
void h2_beam_send_from(h2_bucket_beam *beam, apr_pool_t *p);
/**
* Receive buckets from the beam into the given brigade. Will return APR_EOF
* when reading past an EOS bucket. Reads can be blocking until data is
* available or the beam has been closed. Non-blocking calls return APR_EAGAIN
* if no data is available.
*
* Call from the receiver side only.
*/
apr_status_t h2_beam_receive(h2_bucket_beam *beam,
apr_bucket_brigade *green_buckets,
apr_read_type_e block,
apr_off_t readbytes);
/**
* Determine if beam is empty.
*/
int h2_beam_empty(h2_bucket_beam *beam);
/**
* Determine if beam has handed out proxy buckets that are not destroyed.
*/
int h2_beam_holds_proxies(h2_bucket_beam *beam);
/**
* Abort the beam. Will cleanup any buffered buckets and answer all send
* and receives with APR_ECONNABORTED.
*
* Call from the sender side only.
*/
void h2_beam_abort(h2_bucket_beam *beam);
/**
* Close the beam. Sending an EOS bucket serves the same purpose.
*
* Call from the sender side only.
*/
apr_status_t h2_beam_close(h2_bucket_beam *beam);
/**
* Receives leaves the beam, e.g. will no longer read. This will
* interrupt any sender blocked writing and fail future send.
*
* Call from the receiver side only.
*/
apr_status_t h2_beam_leave(h2_bucket_beam *beam);
int h2_beam_is_closed(h2_bucket_beam *beam);
/**
* Return APR_SUCCESS when all buckets in transit have been handled.
* When called with APR_BLOCK_READ and a mutex set, will wait until the green
* side has consumed all data. Otherwise APR_EAGAIN is returned.
* With clear_buffers set, any queued data is discarded.
* If a timeout is set on the beam, waiting might also time out and
* return APR_ETIMEUP.
*
* Call from the sender side only.
*/
apr_status_t h2_beam_wait_empty(h2_bucket_beam *beam, apr_read_type_e block);
/**
* Set/get the timeout for blocking read/write operations. Only works
* if a mutex has been set for the beam.
*/
void h2_beam_timeout_set(h2_bucket_beam *beam,
apr_interval_time_t timeout);
apr_interval_time_t h2_beam_timeout_get(h2_bucket_beam *beam);
/**
* Set/get the maximum buffer size for beam data (memory footprint).
*/
void h2_beam_buffer_size_set(h2_bucket_beam *beam,
apr_size_t buffer_size);
apr_size_t h2_beam_buffer_size_get(h2_bucket_beam *beam);
/**
* Register a callback to be invoked on the sender side with the
* amount of bytes that have been consumed by the receiver, since the
* last callback invocation or reset.
* @param beam the beam to set the callback on
* @param ev_cb the callback or NULL, called when bytes are consumed
* @param io_cb the callback or NULL, called on sender with bytes consumed
* @param ctx the context to use in callback invocation
*
* Call from the sender side, io callbacks invoked on sender side, ev callback
* from any side.
*/
void h2_beam_on_consumed(h2_bucket_beam *beam,
h2_beam_ev_callback *ev_cb,
h2_beam_io_callback *io_cb, void *ctx);
/**
* Call any registered consumed handler, if any changes have happened
* since the last invocation.
* @return !=0 iff a handler has been called
*
* Needs to be invoked from the sending side.
*/
int h2_beam_report_consumption(h2_bucket_beam *beam);
/**
* Register a callback to be invoked on the receiver side with the
* amount of bytes that have been produces by the sender, since the
* last callback invocation or reset.
* @param beam the beam to set the callback on
* @param io_cb the callback or NULL, called on receiver with bytes produced
* @param ctx the context to use in callback invocation
*
* Call from the receiver side, callbacks invoked on either side.
*/
void h2_beam_on_produced(h2_bucket_beam *beam,
h2_beam_io_callback *io_cb, void *ctx);
/**
* Register a callback that may prevent a file from being beam as
* file handle, forcing the file content to be copied. Then no callback
* is set (NULL), file handles are transferred directly.
* @param beam the beam to set the callback on
* @param io_cb the callback or NULL, called on receiver with bytes produced
* @param ctx the context to use in callback invocation
*
* Call from the receiver side, callbacks invoked on either side.
*/
void h2_beam_on_file_beam(h2_bucket_beam *beam,
h2_beam_can_beam_callback *cb, void *ctx);
/**
* Get the amount of bytes currently buffered in the beam (unread).
*/
apr_off_t h2_beam_get_buffered(h2_bucket_beam *beam);
/**
* Get the memory used by the buffered buckets, approximately.
*/
apr_off_t h2_beam_get_mem_used(h2_bucket_beam *beam);
/**
* Return != 0 iff (some) data from the beam has been received.
*/
int h2_beam_was_received(h2_bucket_beam *beam);
apr_size_t h2_beam_get_files_beamed(h2_bucket_beam *beam);
typedef apr_bucket *h2_bucket_beamer(h2_bucket_beam *beam,
apr_bucket_brigade *dest,
const apr_bucket *src);
void h2_register_bucket_beamer(h2_bucket_beamer *beamer);
void h2_beam_log(h2_bucket_beam *beam, conn_rec *c, int level, const char *msg);
#endif /* h2_bucket_beam_h */