From e364ae8bdbd885fd3262a79d78af5d80689c3ce2 Mon Sep 17 00:00:00 2001 From: Stephan Bosch Date: Wed, 17 Jan 2018 02:48:43 +0100 Subject: [PATCH] lib: istream: Allow switching to a specific ioloop. --- src/lib-http/http-server-request.c | 6 ++++-- src/lib/istream-multiplex.c | 7 ++++--- src/lib/istream-private.h | 3 ++- src/lib/istream-timeout.c | 19 ++++++++++++------- src/lib/istream.c | 16 +++++++++++----- src/lib/istream.h | 5 ++++- 6 files changed, 37 insertions(+), 19 deletions(-) diff --git a/src/lib-http/http-server-request.c b/src/lib-http/http-server-request.c index a332feed8e..c922bb42d8 100644 --- a/src/lib-http/http-server-request.c +++ b/src/lib-http/http-server-request.c @@ -461,7 +461,8 @@ struct http_server_istream { }; static void -http_server_istream_switch_ioloop(struct istream_private *stream) +http_server_istream_switch_ioloop_to(struct istream_private *stream, + struct ioloop *ioloop) { struct http_server_istream *hsristream = (struct http_server_istream *)stream; @@ -469,6 +470,7 @@ http_server_istream_switch_ioloop(struct istream_private *stream) if (hsristream->istream.istream.blocking) return; + i_assert(ioloop == current_ioloop); http_server_connection_switch_ioloop(hsristream->req->conn); } @@ -581,7 +583,7 @@ http_server_request_get_payload_input(struct http_server_request *req, hsristream->istream.stream_size_passthrough = TRUE; hsristream->istream.read = http_server_istream_read; - hsristream->istream.switch_ioloop = http_server_istream_switch_ioloop; + hsristream->istream.switch_ioloop_to = http_server_istream_switch_ioloop_to; hsristream->istream.iostream.destroy = http_server_istream_destroy; hsristream->istream.istream.readable_fd = FALSE; diff --git a/src/lib/istream-multiplex.c b/src/lib/istream-multiplex.c index ec024d4deb..13eae8c3ca 100644 --- a/src/lib/istream-multiplex.c +++ b/src/lib/istream-multiplex.c @@ -189,11 +189,12 @@ static ssize_t i_stream_multiplex_ichannel_read(struct istream_private *stream) } static void -i_stream_multiplex_ichannel_switch_ioloop(struct istream_private *stream) +i_stream_multiplex_ichannel_switch_ioloop_to(struct istream_private *stream, + struct ioloop *ioloop) { struct multiplex_ichannel *channel = (struct multiplex_ichannel*)stream; - i_stream_switch_ioloop(channel->mstream->parent); + i_stream_switch_ioloop_to(channel->mstream->parent, ioloop); } static void @@ -244,7 +245,7 @@ i_stream_add_channel_real(struct multiplex_istream *mstream, uint8_t cid) channel->cid = cid; channel->mstream = mstream; channel->istream.read = i_stream_multiplex_ichannel_read; - channel->istream.switch_ioloop = i_stream_multiplex_ichannel_switch_ioloop; + channel->istream.switch_ioloop_to = i_stream_multiplex_ichannel_switch_ioloop_to; channel->istream.iostream.close = i_stream_multiplex_ichannel_close; channel->istream.iostream.destroy = i_stream_multiplex_ichannel_destroy; channel->istream.max_buffer_size = mstream->bufsize; diff --git a/src/lib/istream-private.h b/src/lib/istream-private.h index d2304b9c20..a88653133d 100644 --- a/src/lib/istream-private.h +++ b/src/lib/istream-private.h @@ -19,7 +19,8 @@ struct istream_private { void (*sync)(struct istream_private *stream); int (*stat)(struct istream_private *stream, bool exact); int (*get_size)(struct istream_private *stream, bool exact, uoff_t *size_r); - void (*switch_ioloop)(struct istream_private *stream); + void (*switch_ioloop_to)(struct istream_private *stream, + struct ioloop *ioloop); struct istream_snapshot * (*snapshot)(struct istream_private *stream, struct istream_snapshot *prev_snapshot); diff --git a/src/lib/istream-timeout.c b/src/lib/istream-timeout.c index a7b1fd6230..d7d8f98bb4 100644 --- a/src/lib/istream-timeout.c +++ b/src/lib/istream-timeout.c @@ -27,16 +27,18 @@ static void i_stream_timeout_close(struct iostream_private *stream, i_stream_close(tstream->istream.parent); } -static void i_stream_timeout_switch_ioloop(struct istream_private *stream) +static void i_stream_timeout_switch_ioloop_to(struct istream_private *stream, + struct ioloop *ioloop) { struct timeout_istream *tstream = (struct timeout_istream *)stream; if (tstream->to != NULL) - tstream->to = io_loop_move_timeout(&tstream->to); + tstream->to = io_loop_move_timeout_to(ioloop, &tstream->to); } static void i_stream_timeout(struct timeout_istream *tstream) { + struct iostream_private *iostream = &tstream->istream.iostream; unsigned int over_msecs; int diff; @@ -54,8 +56,9 @@ static void i_stream_timeout(struct timeout_istream *tstream) /* we haven't reached the read timeout yet, update it */ if (diff < 0) diff = 0; - tstream->to = timeout_add(tstream->timeout_msecs - diff, - i_stream_timeout, tstream); + tstream->to = timeout_add_to(io_stream_get_ioloop(iostream), + tstream->timeout_msecs - diff, + i_stream_timeout, tstream); return; } over_msecs = diff - tstream->timeout_msecs; @@ -86,6 +89,7 @@ static ssize_t i_stream_timeout_read(struct istream_private *stream) { struct timeout_istream *tstream = (struct timeout_istream *)stream; + struct iostream_private *iostream = &tstream->istream.iostream; ssize_t ret; i_stream_seek(stream->parent, stream->parent_start_offset + @@ -105,8 +109,9 @@ i_stream_timeout_read(struct istream_private *stream) /* first read. add the timeout here instead of in init in case the stream is created long before it's actually read from. */ - tstream->to = timeout_add(tstream->timeout_msecs, - i_stream_timeout, tstream); + tstream->to = timeout_add_to(io_stream_get_ioloop(iostream), + tstream->timeout_msecs, + i_stream_timeout, tstream); i_stream_timeout_set_pending(tstream); } else if (ret > 0 && tstream->to != NULL) { /* we read something, reset the timeout */ @@ -131,7 +136,7 @@ i_stream_create_timeout(struct istream *input, unsigned int timeout_msecs) tstream->created = ioloop_time; tstream->istream.read = i_stream_timeout_read; - tstream->istream.switch_ioloop = i_stream_timeout_switch_ioloop; + tstream->istream.switch_ioloop_to = i_stream_timeout_switch_ioloop_to; tstream->istream.iostream.close = i_stream_timeout_close; tstream->istream.istream.readable_fd = input->readable_fd; diff --git a/src/lib/istream.c b/src/lib/istream.c index 1486f24dbf..bb77280479 100644 --- a/src/lib/istream.c +++ b/src/lib/istream.c @@ -926,18 +926,24 @@ void i_stream_set_input_pending(struct istream *stream, bool pending) io_set_pending(stream->real_stream->io); } -void i_stream_switch_ioloop(struct istream *stream) +void i_stream_switch_ioloop_to(struct istream *stream, struct ioloop *ioloop) { - io_stream_switch_ioloop_to(&stream->real_stream->iostream, - current_ioloop); + io_stream_switch_ioloop_to(&stream->real_stream->iostream, ioloop); do { - if (stream->real_stream->switch_ioloop != NULL) - stream->real_stream->switch_ioloop(stream->real_stream); + if (stream->real_stream->switch_ioloop_to != NULL) { + stream->real_stream->switch_ioloop_to( + stream->real_stream, ioloop); + } stream = stream->real_stream->parent; } while (stream != NULL); } +void i_stream_switch_ioloop(struct istream *stream) +{ + i_stream_switch_ioloop_to(stream, current_ioloop); +} + void i_stream_set_io(struct istream *stream, struct io *io) { stream = i_stream_get_root_io(stream); diff --git a/src/lib/istream.h b/src/lib/istream.h index 3c99bff641..f9ccf94b86 100644 --- a/src/lib/istream.h +++ b/src/lib/istream.h @@ -4,6 +4,8 @@ /* Note that some systems (Solaris) may use a macro to redefine struct stat */ #include +struct ioloop; + struct istream { uoff_t v_offset; @@ -233,7 +235,8 @@ bool i_stream_add_data(struct istream *stream, const unsigned char *data, void i_stream_set_input_pending(struct istream *stream, bool pending); /* If there are any I/O loop items associated with the stream, move all of - them to current_ioloop. */ + them to provided/current ioloop. */ +void i_stream_switch_ioloop_to(struct istream *stream, struct ioloop *ioloop); void i_stream_switch_ioloop(struct istream *stream); #endif