You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kudu.apache.org by Michael Ho <kw...@cloudera.com> on 2017/10/13 16:37:44 UTC
RFC: Canceling RPC in mid-transmission
There were some recent discussions about the proposed mechanism to cancel
an RPC in mid-transmission. I am writing this email to summarize the
problem cancellation is trying to solve and some proposed solutions.
As part of IMPALA-2567 <https://issues.apache.org/jira/browse/IMPALA-2567>,
Impala is porting its RPC facility to utilize the Kudu RPC library. One
major feature of KRPC is asynchronous RPC. This prevents the need for
Impala to create a separate thread for each communication channel between
fragment instances running on different Impala backends. When an
asynchronous RPC is in progress, the KRPC code needs to hold on to the RPC
payload provided and owned by a fragment instance. We have a singleton
messenger instance per Impalad node. The connection between two hosts is
shared by different fragment instances.
At any point when the RPC is in-progress, the coordinator can cancel all
fragment instances of a query for various reasons. In which case, the
fragment execution threads need to cancel their in-flight RPC and free up
the RPC payloads (e.g. which are passed as sidecars of the RPC request)
within a reasonable amount of time (e.g. 500ms). Currently in the KRPC
code, when a cancellation is requested when an outbound call is in SENDING
state (i.e. it has potentially sent some or the whole header to the remote
server), it will wait for the entire payload to be sent before invoking the
completion callback specified in the async RPC request. At which point, the
KRPC client can assume that the RPC payload is no longer referenced by the
KRPC code and it can be freed.
When the network connections between Impalad backends is functioning well,
it shouldn't be long before the payload is done sending as Impalad
internally tries to bound the payload size to 8MB (but not necessarily
guaranteed in some degenerated case e.g. row with 1GB string). However,
when the network gets congested or is very lossy, there is in theory no
bound on how long it takes to finish sending the payload. In other words,
the cancellation of the RPC can take unbounded amount of time to complete
and thus the cancellation of a fragment instance can also take infinitely
long. This is a chicken-and-egg problem. Usually, the user cancels the
query because the poor network performance caused it to run for too long.
With the unbounded wait, the user cannot even cancel the query within a
bounded period of time.
Similar problem arises when an RPC times out. As the KRPC code stands
today, there is no guarantee that the payload is not referenced by the KRPC
code when the completion callback is invoked due to timeout. The client
doesn't get any notification on when the payload can be freed. In theory,
this can be solved by deferring the callback until the payload has been
completely sent. However, this may not always be desirable due to the
unbounded wait mentioned above. For example, if a client requests a timeout
of 60s, it's not desirable to invoke the callback say 300s after the
timeout.
If you agree that the above are indeed problems, we need a RPC cancellation
mechanism which meets the following requirements:
- the RPC payload is no longer referenced by the KRPC code when the
completion callback is invoked.
- the time between the cancellation request and the invocation of the
completion callback should be reasonably bounded.
The followings are two different proposals floated:
1. Upgrade the KRPC protocol to insert a footer at the very end of each
outbound call transfer (i.e. after header, serialized protobuf request,
sidecars etc). The footer contains a flag which is parsed by the server to
determine whether the inbound call should be discarded. Normally, this flag
will be cleared. When an outbound call is cancelled while it's still in the
SENDING state, the footer is modified in place to have the flag set. In
addition, the sidecars of the outbound call transfers are modified to point
to a dummy buffer. In this way, the slices in an outbound transfer no
longer references the original payload of the RPC and the footer guarantees
that whatever is received by the server will be discarded so it's okay to
replace the original payload with dummy bytes. This allows the completion
callback to be invoked without waiting for the entire payload to be sent.
Pros:
- relatively low overhead in network consumption (1 byte per outbound call
transfer)
- cancellation is complete once the reactor thread handles the cancellation
event and is not reliant on the network condition.
Cons:
- the remaining bytes of the payload still need to be sent across a
congested network after cancellation.
- added complexity to the protocol.
The details of the proposal is in KUDU-2065
<https://issues.apache.org/jira/browse/KUDU-2065>. The patch which
implemented this idea is at https://gerrit.cloudera.org/#/c/7599/
It was later reverted due to KUDU-2110
<https://issues.apache.org/jira/browse/KUDU-2110>.
2. Keep the KRPC protocol unchanged. When an outbound call is cancelled
when it's in SENDING state, register a timer with a timeout (e.g. 500ms).
When the timer expires, invoke a callback to call shutdown(fd, SHUT_WR) on
the socket. This shuts down one-half of the connection and prevents more
bytes from being sent while allowing bytes to be read on the client side.
recv() will eventually return 0 on the server side. After all other
in-flight RPCs for that connection have been replied to, the server should
close the connection. The client, after reading all the replies, should
also destroy the Connection object. All RPCs behind the cancelled the RPC
in the outbound queue on the client side are aborted and need to be retried.
Pro:
- stop piling on a congested network after cancelling an RPC.
- cancellation is still bounded in time.
Con:
- For Impalad, each host may be connected to all other hosts in the
cluster. In the worst case, this means closing all n connections. In total,
that could be n^2 connection being re-established concurrently. This can
easily trigger negotiation timeout in a secure cluster based on our testing
on bolt1k cluster, leading to unpredictable failures.
- The RPCs behind the cancelled RPC in the outbound queue cannot be retried
until all previous in-flight RPCs have been replied to. This can be
unboundedly long if the remote server is overloaded.
Please feel free to comment on the proposals above or suggest new ideas.
--
Thanks,
Michael
Re: RFC: Canceling RPC in mid-transmission
Posted by Michael Ho <kw...@cloudera.com>.
It's true that there is no timely shutdown in the presence of full network
partitions but most likely statestore will detect that some hosts are
unreachable and eventually fail a bunch a queries.
However, even without any network partition, the shutdown of query
fragments on many hosts may not happen in a timely manner if RPC
cancellation is stuck waiting for a few nodes overloaded with network
traffic.
This is a case in which we don't want to regress from our existing
implementation with thrift RPC.
On Fri, Oct 13, 2017 at 2:12 PM, Dan Burkert <da...@apache.org> wrote:
> Given that Impala can't guarantee timely shutdown in the presence of full
> network partitions, and it only takes 130Mbps of bandwidth to transmit 8MiB
> in 500ms, why are changes to the protocol or lifecycle of RPCs necessary?
>
> - Dan
>
> On Fri, Oct 13, 2017 at 1:14 PM, Michael Ho <kw...@cloudera.com> wrote:
>
>> And to provide more context for Impala's query cancellation, the
>> cancellation is asynchronous. In other words, the coordinator is only
>> responsible for notifying the remote Impalad to shut down the fragment
>> instances at a convenient time. The coordinator will not wait for
>> acknowledgement from all Impalad before proceeding. The assumption is that
>> the executor threads will periodically poll for cancellation request and
>> free up the resources soon after it sees one.
>>
>> On Fri, Oct 13, 2017 at 1:10 PM, Michael Ho <kw...@cloudera.com> wrote:
>>
>>> The coordinator RPC to cancel the fragment instance usually has much
>>> smaller payload compared to the RPC for transmitting data between Impalad
>>> nodes so it has a higher chance of coming through. I agree that, when the
>>> network is in bad shape, cancellation requests may not reach all Impalad
>>> nodes. However, keep in mind that it's not necessarily the case that the
>>> entire network is swamped. It can be that one or a few nodes in the cluster
>>> which were pounded on with a lot of network traffic. In those cases, it
>>> seems all other fragment instances on other Impalad nodes will still block
>>> waiting for these few overloaded during cancellation and hold up memory
>>> resources on those Impalad nodes.
>>>
>>> Thanks,
>>> Michael
>>>
>>> On Fri, Oct 13, 2017 at 11:30 AM, Dan Burkert <da...@cloudera.com> wrote:
>>>
>>>> I'm still hung up on the premise that if the network is down,
>>>> cancellation should still be expected to complete in bounded time. If the
>>>> network is partitioned or congested to the point of non-delivery, how is
>>>> the coordinator going to deliver the cancellation messages to the fragments
>>>> in the first place? If we're allowing for arbitrary network delays, there
>>>> is no way to make this guarantee. We're laser focussed on this issue of
>>>> cancelling mid-transmission RPCs, but transmission is a very minor slice of
>>>> the whole RPC lifecycle, and I don't believe it's worth complicating the
>>>> RPC protocol or lifecycle over.
>>>>
>>>> - Dan
>>>>
>>>> On Fri, Oct 13, 2017 at 11:01 AM, Sailesh Mukil <sa...@cloudera.com>
>>>> wrote:
>>>>
>>>>> On Fri, Oct 13, 2017 at 9:37 AM, Michael Ho <kw...@cloudera.com> wrote:
>>>>>
>>>>> > There were some recent discussions about the proposed mechanism to
>>>>> cancel
>>>>> > an RPC in mid-transmission. I am writing this email to summarize the
>>>>> > problem cancellation is trying to solve and some proposed solutions.
>>>>> >
>>>>> > As part of IMPALA-2567 <https://issues.apache.org/jir
>>>>> a/browse/IMPALA-2567>,
>>>>>
>>>>> > Impala is porting its RPC facility to utilize the Kudu RPC library.
>>>>> One
>>>>> > major feature of KRPC is asynchronous RPC. This prevents the need for
>>>>> > Impala to create a separate thread for each communication channel
>>>>> between
>>>>> > fragment instances running on different Impala backends. When an
>>>>> > asynchronous RPC is in progress, the KRPC code needs to hold on to
>>>>> the RPC
>>>>> > payload provided and owned by a fragment instance. We have a
>>>>> singleton
>>>>> > messenger instance per Impalad node. The connection between two
>>>>> hosts is
>>>>> > shared by different fragment instances.
>>>>> >
>>>>> > At any point when the RPC is in-progress, the coordinator can cancel
>>>>> all
>>>>> > fragment instances of a query for various reasons. In which case, the
>>>>> > fragment execution threads need to cancel their in-flight RPC and
>>>>> free up
>>>>> > the RPC payloads (e.g. which are passed as sidecars of the RPC
>>>>> request)
>>>>> > within a reasonable amount of time (e.g. 500ms). Currently in the
>>>>> KRPC
>>>>> > code, when a cancellation is requested when an outbound call is in
>>>>> SENDING
>>>>> > state (i.e. it has potentially sent some or the whole header to the
>>>>> remote
>>>>> > server), it will wait for the entire payload to be sent before
>>>>> invoking the
>>>>> > completion callback specified in the async RPC request. At which
>>>>> point, the
>>>>> > KRPC client can assume that the RPC payload is no longer referenced
>>>>> by the
>>>>> > KRPC code and it can be freed.
>>>>> >
>>>>> > When the network connections between Impalad backends is functioning
>>>>> well,
>>>>> > it shouldn't be long before the payload is done sending as Impalad
>>>>> > internally tries to bound the payload size to 8MB (but not
>>>>> necessarily
>>>>> > guaranteed in some degenerated case e.g. row with 1GB string).
>>>>> However,
>>>>> > when the network gets congested or is very lossy, there is in theory
>>>>> no
>>>>> > bound on how long it takes to finish sending the payload. In other
>>>>> words,
>>>>> > the cancellation of the RPC can take unbounded amount of time to
>>>>> complete
>>>>> > and thus the cancellation of a fragment instance can also take
>>>>> infinitely
>>>>> > long. This is a chicken-and-egg problem. Usually, the user cancels
>>>>> the
>>>>> > query because the poor network performance caused it to run for too
>>>>> long.
>>>>> > With the unbounded wait, the user cannot even cancel the query
>>>>> within a
>>>>> > bounded period of time.
>>>>> >
>>>>> > Similar problem arises when an RPC times out. As the KRPC code stands
>>>>> > today, there is no guarantee that the payload is not referenced by
>>>>> the KRPC
>>>>> > code when the completion callback is invoked due to timeout. The
>>>>> client
>>>>> > doesn't get any notification on when the payload can be freed. In
>>>>> theory,
>>>>> > this can be solved by deferring the callback until the payload has
>>>>> been
>>>>> > completely sent. However, this may not always be desirable due to the
>>>>> > unbounded wait mentioned above. For example, if a client requests a
>>>>> timeout
>>>>> > of 60s, it's not desirable to invoke the callback say 300s after the
>>>>> > timeout.
>>>>> >
>>>>> > If you agree that the above are indeed problems, we need a RPC
>>>>> > cancellation mechanism which meets the following requirements:
>>>>> >
>>>>> > - the RPC payload is no longer referenced by the KRPC code when the
>>>>> > completion callback is invoked.
>>>>> > - the time between the cancellation request and the invocation of the
>>>>> > completion callback should be reasonably bounded.
>>>>> >
>>>>> > The followings are two different proposals floated:
>>>>> >
>>>>> > 1. Upgrade the KRPC protocol to insert a footer at the very end of
>>>>> each
>>>>> > outbound call transfer (i.e. after header, serialized protobuf
>>>>> request,
>>>>> > sidecars etc). The footer contains a flag which is parsed by the
>>>>> server to
>>>>> > determine whether the inbound call should be discarded. Normally,
>>>>> this flag
>>>>> > will be cleared. When an outbound call is cancelled while it's still
>>>>> in the
>>>>> > SENDING state, the footer is modified in place to have the flag set.
>>>>> In
>>>>> > addition, the sidecars of the outbound call transfers are modified
>>>>> to point
>>>>> > to a dummy buffer. In this way, the slices in an outbound transfer no
>>>>> > longer references the original payload of the RPC and the footer
>>>>> guarantees
>>>>> > that whatever is received by the server will be discarded so it's
>>>>> okay to
>>>>> > replace the original payload with dummy bytes. This allows the
>>>>> completion
>>>>> > callback to be invoked without waiting for the entire payload to be
>>>>> sent.
>>>>> >
>>>>> > Pros:
>>>>> > - relatively low overhead in network consumption (1 byte per
>>>>> outbound call
>>>>> > transfer)
>>>>> > - cancellation is complete once the reactor thread handles the
>>>>> > cancellation event and is not reliant on the network condition.
>>>>> >
>>>>> > Cons:
>>>>> > - the remaining bytes of the payload still need to be sent across a
>>>>> > congested network after cancellation.
>>>>> > - added complexity to the protocol.
>>>>> >
>>>>> > The details of the proposal is in KUDU-2065
>>>>> > <https://issues.apache.org/jira/browse/KUDU-2065>. The patch which
>>>>> > implemented this idea is at https://gerrit.cloudera.org/#/c/7599/
>>>>> > It was later reverted due to KUDU-2110
>>>>> > <https://issues.apache.org/jira/browse/KUDU-2110>.
>>>>>
>>>>> >
>>>>> > 2. Keep the KRPC protocol unchanged. When an outbound call is
>>>>> cancelled
>>>>> > when it's in SENDING state, register a timer with a timeout (e.g.
>>>>> 500ms).
>>>>> > When the timer expires, invoke a callback to call shutdown(fd,
>>>>> SHUT_WR) on
>>>>> > the socket. This shuts down one-half of the connection and prevents
>>>>> more
>>>>> > bytes from being sent while allowing bytes to be read on the client
>>>>> side.
>>>>> > recv() will eventually return 0 on the server side. After all other
>>>>> > in-flight RPCs for that connection have been replied to, the server
>>>>> should
>>>>> > close the connection. The client, after reading all the replies,
>>>>> should
>>>>> > also destroy the Connection object. All RPCs behind the cancelled
>>>>> the RPC
>>>>> > in the outbound queue on the client side are aborted and need to be
>>>>> retried.
>>>>> >
>>>>> > Pro:
>>>>> > - stop piling on a congested network after cancelling an RPC.
>>>>> > - cancellation is still bounded in time.
>>>>> >
>>>>> > Con:
>>>>> > - For Impalad, each host may be connected to all other hosts in the
>>>>> > cluster. In the worst case, this means closing all n connections. In
>>>>> total,
>>>>> > that could be n^2 connection being re-established concurrently. This
>>>>> can
>>>>> > easily trigger negotiation timeout in a secure cluster based on our
>>>>> testing
>>>>> > on bolt1k cluster, leading to unpredictable failures.
>>>>> >
>>>>> > - The RPCs behind the cancelled RPC in the outbound queue cannot be
>>>>> > retried until all previous in-flight RPCs have been replied to. This
>>>>> can be
>>>>> > unboundedly long if the remote server is overloaded.
>>>>> >
>>>>> >
>>>>> Regarding option 2, although this may work well with certain
>>>>> workloads, I'm
>>>>> not sure that this would be the right way to go about it. Also, in some
>>>>> scenarios, it could be very slow.
>>>>>
>>>>> From the viewpoint of Impala, cancellation can happen in many
>>>>> scenarios.
>>>>> Other than a query being cancelled by a user, cancellations also happen
>>>>> when there are runtime errors (OOM, found a bad file in the storage
>>>>> layer,
>>>>> etc.)
>>>>>
>>>>> Even aside from the negotiation storm mentioned above by Michael, if we
>>>>> were to tear down the connection for every cancel request we get, it
>>>>> will
>>>>> cause a slow down for all other queries running on those nodes as well,
>>>>> since they have to now wait longer until the connection is re-setup.
>>>>>
>>>>> We can imagine a scenario where the cluster is under high stress
>>>>> leading
>>>>> quite a few queries to OOM, which would send out cancellations to all
>>>>> in
>>>>> flight RPCs for those respective queries, causing a chain of
>>>>> cancellations.
>>>>> We can also imagine multiple CANCEL requests from different queries
>>>>> being
>>>>> queued for the same connection, so the first CANCEL request tears down
>>>>> the
>>>>> connection, and sets it up again, only to be torn down again by another
>>>>> queued CANCEL request. Sure, we can try to detect multiple CANCEL
>>>>> requests
>>>>> in a queue and do it only once, but that goes back to adding
>>>>> complexity to
>>>>> the protocol.
>>>>>
>>>>> To my knowledge, I don't know of a RPC library that takes option 2,
>>>>> but I
>>>>> know at least one that does something similar to option 1 (grpc).
>>>>> That's my
>>>>> two cents on the issue, but I may have missed out thinking about it
>>>>> from
>>>>> different perspectives, so feel free to correct me if I got something
>>>>> wrong.
>>>>>
>>>>> Please feel free to comment on the proposals above or suggest new
>>>>> ideas.
>>>>> >
>>>>> > --
>>>>> > Thanks,
>>>>> > Michael
>>>>> >
>>>>>
>>>>
>>>>
>>>
>>>
>>> --
>>> Thanks,
>>> Michael
>>>
>>
>>
>>
>> --
>> Thanks,
>> Michael
>>
>
>
--
Thanks,
Michael
Re: RFC: Canceling RPC in mid-transmission
Posted by Dan Burkert <da...@apache.org>.
Given that Impala can't guarantee timely shutdown in the presence of full
network partitions, and it only takes 130Mbps of bandwidth to transmit 8MiB
in 500ms, why are changes to the protocol or lifecycle of RPCs necessary?
- Dan
On Fri, Oct 13, 2017 at 1:14 PM, Michael Ho <kw...@cloudera.com> wrote:
> And to provide more context for Impala's query cancellation, the
> cancellation is asynchronous. In other words, the coordinator is only
> responsible for notifying the remote Impalad to shut down the fragment
> instances at a convenient time. The coordinator will not wait for
> acknowledgement from all Impalad before proceeding. The assumption is that
> the executor threads will periodically poll for cancellation request and
> free up the resources soon after it sees one.
>
> On Fri, Oct 13, 2017 at 1:10 PM, Michael Ho <kw...@cloudera.com> wrote:
>
>> The coordinator RPC to cancel the fragment instance usually has much
>> smaller payload compared to the RPC for transmitting data between Impalad
>> nodes so it has a higher chance of coming through. I agree that, when the
>> network is in bad shape, cancellation requests may not reach all Impalad
>> nodes. However, keep in mind that it's not necessarily the case that the
>> entire network is swamped. It can be that one or a few nodes in the cluster
>> which were pounded on with a lot of network traffic. In those cases, it
>> seems all other fragment instances on other Impalad nodes will still block
>> waiting for these few overloaded during cancellation and hold up memory
>> resources on those Impalad nodes.
>>
>> Thanks,
>> Michael
>>
>> On Fri, Oct 13, 2017 at 11:30 AM, Dan Burkert <da...@cloudera.com> wrote:
>>
>>> I'm still hung up on the premise that if the network is down,
>>> cancellation should still be expected to complete in bounded time. If the
>>> network is partitioned or congested to the point of non-delivery, how is
>>> the coordinator going to deliver the cancellation messages to the fragments
>>> in the first place? If we're allowing for arbitrary network delays, there
>>> is no way to make this guarantee. We're laser focussed on this issue of
>>> cancelling mid-transmission RPCs, but transmission is a very minor slice of
>>> the whole RPC lifecycle, and I don't believe it's worth complicating the
>>> RPC protocol or lifecycle over.
>>>
>>> - Dan
>>>
>>> On Fri, Oct 13, 2017 at 11:01 AM, Sailesh Mukil <sa...@cloudera.com>
>>> wrote:
>>>
>>>> On Fri, Oct 13, 2017 at 9:37 AM, Michael Ho <kw...@cloudera.com> wrote:
>>>>
>>>> > There were some recent discussions about the proposed mechanism to
>>>> cancel
>>>> > an RPC in mid-transmission. I am writing this email to summarize the
>>>> > problem cancellation is trying to solve and some proposed solutions.
>>>> >
>>>> > As part of IMPALA-2567 <https://issues.apache.org/jir
>>>> a/browse/IMPALA-2567>,
>>>>
>>>> > Impala is porting its RPC facility to utilize the Kudu RPC library.
>>>> One
>>>> > major feature of KRPC is asynchronous RPC. This prevents the need for
>>>> > Impala to create a separate thread for each communication channel
>>>> between
>>>> > fragment instances running on different Impala backends. When an
>>>> > asynchronous RPC is in progress, the KRPC code needs to hold on to
>>>> the RPC
>>>> > payload provided and owned by a fragment instance. We have a singleton
>>>> > messenger instance per Impalad node. The connection between two hosts
>>>> is
>>>> > shared by different fragment instances.
>>>> >
>>>> > At any point when the RPC is in-progress, the coordinator can cancel
>>>> all
>>>> > fragment instances of a query for various reasons. In which case, the
>>>> > fragment execution threads need to cancel their in-flight RPC and
>>>> free up
>>>> > the RPC payloads (e.g. which are passed as sidecars of the RPC
>>>> request)
>>>> > within a reasonable amount of time (e.g. 500ms). Currently in the KRPC
>>>> > code, when a cancellation is requested when an outbound call is in
>>>> SENDING
>>>> > state (i.e. it has potentially sent some or the whole header to the
>>>> remote
>>>> > server), it will wait for the entire payload to be sent before
>>>> invoking the
>>>> > completion callback specified in the async RPC request. At which
>>>> point, the
>>>> > KRPC client can assume that the RPC payload is no longer referenced
>>>> by the
>>>> > KRPC code and it can be freed.
>>>> >
>>>> > When the network connections between Impalad backends is functioning
>>>> well,
>>>> > it shouldn't be long before the payload is done sending as Impalad
>>>> > internally tries to bound the payload size to 8MB (but not necessarily
>>>> > guaranteed in some degenerated case e.g. row with 1GB string).
>>>> However,
>>>> > when the network gets congested or is very lossy, there is in theory
>>>> no
>>>> > bound on how long it takes to finish sending the payload. In other
>>>> words,
>>>> > the cancellation of the RPC can take unbounded amount of time to
>>>> complete
>>>> > and thus the cancellation of a fragment instance can also take
>>>> infinitely
>>>> > long. This is a chicken-and-egg problem. Usually, the user cancels the
>>>> > query because the poor network performance caused it to run for too
>>>> long.
>>>> > With the unbounded wait, the user cannot even cancel the query within
>>>> a
>>>> > bounded period of time.
>>>> >
>>>> > Similar problem arises when an RPC times out. As the KRPC code stands
>>>> > today, there is no guarantee that the payload is not referenced by
>>>> the KRPC
>>>> > code when the completion callback is invoked due to timeout. The
>>>> client
>>>> > doesn't get any notification on when the payload can be freed. In
>>>> theory,
>>>> > this can be solved by deferring the callback until the payload has
>>>> been
>>>> > completely sent. However, this may not always be desirable due to the
>>>> > unbounded wait mentioned above. For example, if a client requests a
>>>> timeout
>>>> > of 60s, it's not desirable to invoke the callback say 300s after the
>>>> > timeout.
>>>> >
>>>> > If you agree that the above are indeed problems, we need a RPC
>>>> > cancellation mechanism which meets the following requirements:
>>>> >
>>>> > - the RPC payload is no longer referenced by the KRPC code when the
>>>> > completion callback is invoked.
>>>> > - the time between the cancellation request and the invocation of the
>>>> > completion callback should be reasonably bounded.
>>>> >
>>>> > The followings are two different proposals floated:
>>>> >
>>>> > 1. Upgrade the KRPC protocol to insert a footer at the very end of
>>>> each
>>>> > outbound call transfer (i.e. after header, serialized protobuf
>>>> request,
>>>> > sidecars etc). The footer contains a flag which is parsed by the
>>>> server to
>>>> > determine whether the inbound call should be discarded. Normally,
>>>> this flag
>>>> > will be cleared. When an outbound call is cancelled while it's still
>>>> in the
>>>> > SENDING state, the footer is modified in place to have the flag set.
>>>> In
>>>> > addition, the sidecars of the outbound call transfers are modified to
>>>> point
>>>> > to a dummy buffer. In this way, the slices in an outbound transfer no
>>>> > longer references the original payload of the RPC and the footer
>>>> guarantees
>>>> > that whatever is received by the server will be discarded so it's
>>>> okay to
>>>> > replace the original payload with dummy bytes. This allows the
>>>> completion
>>>> > callback to be invoked without waiting for the entire payload to be
>>>> sent.
>>>> >
>>>> > Pros:
>>>> > - relatively low overhead in network consumption (1 byte per outbound
>>>> call
>>>> > transfer)
>>>> > - cancellation is complete once the reactor thread handles the
>>>> > cancellation event and is not reliant on the network condition.
>>>> >
>>>> > Cons:
>>>> > - the remaining bytes of the payload still need to be sent across a
>>>> > congested network after cancellation.
>>>> > - added complexity to the protocol.
>>>> >
>>>> > The details of the proposal is in KUDU-2065
>>>> > <https://issues.apache.org/jira/browse/KUDU-2065>. The patch which
>>>> > implemented this idea is at https://gerrit.cloudera.org/#/c/7599/
>>>> > It was later reverted due to KUDU-2110
>>>> > <https://issues.apache.org/jira/browse/KUDU-2110>.
>>>>
>>>> >
>>>> > 2. Keep the KRPC protocol unchanged. When an outbound call is
>>>> cancelled
>>>> > when it's in SENDING state, register a timer with a timeout (e.g.
>>>> 500ms).
>>>> > When the timer expires, invoke a callback to call shutdown(fd,
>>>> SHUT_WR) on
>>>> > the socket. This shuts down one-half of the connection and prevents
>>>> more
>>>> > bytes from being sent while allowing bytes to be read on the client
>>>> side.
>>>> > recv() will eventually return 0 on the server side. After all other
>>>> > in-flight RPCs for that connection have been replied to, the server
>>>> should
>>>> > close the connection. The client, after reading all the replies,
>>>> should
>>>> > also destroy the Connection object. All RPCs behind the cancelled the
>>>> RPC
>>>> > in the outbound queue on the client side are aborted and need to be
>>>> retried.
>>>> >
>>>> > Pro:
>>>> > - stop piling on a congested network after cancelling an RPC.
>>>> > - cancellation is still bounded in time.
>>>> >
>>>> > Con:
>>>> > - For Impalad, each host may be connected to all other hosts in the
>>>> > cluster. In the worst case, this means closing all n connections. In
>>>> total,
>>>> > that could be n^2 connection being re-established concurrently. This
>>>> can
>>>> > easily trigger negotiation timeout in a secure cluster based on our
>>>> testing
>>>> > on bolt1k cluster, leading to unpredictable failures.
>>>> >
>>>> > - The RPCs behind the cancelled RPC in the outbound queue cannot be
>>>> > retried until all previous in-flight RPCs have been replied to. This
>>>> can be
>>>> > unboundedly long if the remote server is overloaded.
>>>> >
>>>> >
>>>> Regarding option 2, although this may work well with certain workloads,
>>>> I'm
>>>> not sure that this would be the right way to go about it. Also, in some
>>>> scenarios, it could be very slow.
>>>>
>>>> From the viewpoint of Impala, cancellation can happen in many scenarios.
>>>> Other than a query being cancelled by a user, cancellations also happen
>>>> when there are runtime errors (OOM, found a bad file in the storage
>>>> layer,
>>>> etc.)
>>>>
>>>> Even aside from the negotiation storm mentioned above by Michael, if we
>>>> were to tear down the connection for every cancel request we get, it
>>>> will
>>>> cause a slow down for all other queries running on those nodes as well,
>>>> since they have to now wait longer until the connection is re-setup.
>>>>
>>>> We can imagine a scenario where the cluster is under high stress leading
>>>> quite a few queries to OOM, which would send out cancellations to all in
>>>> flight RPCs for those respective queries, causing a chain of
>>>> cancellations.
>>>> We can also imagine multiple CANCEL requests from different queries
>>>> being
>>>> queued for the same connection, so the first CANCEL request tears down
>>>> the
>>>> connection, and sets it up again, only to be torn down again by another
>>>> queued CANCEL request. Sure, we can try to detect multiple CANCEL
>>>> requests
>>>> in a queue and do it only once, but that goes back to adding complexity
>>>> to
>>>> the protocol.
>>>>
>>>> To my knowledge, I don't know of a RPC library that takes option 2, but
>>>> I
>>>> know at least one that does something similar to option 1 (grpc).
>>>> That's my
>>>> two cents on the issue, but I may have missed out thinking about it from
>>>> different perspectives, so feel free to correct me if I got something
>>>> wrong.
>>>>
>>>> Please feel free to comment on the proposals above or suggest new ideas.
>>>> >
>>>> > --
>>>> > Thanks,
>>>> > Michael
>>>> >
>>>>
>>>
>>>
>>
>>
>> --
>> Thanks,
>> Michael
>>
>
>
>
> --
> Thanks,
> Michael
>
Re: RFC: Canceling RPC in mid-transmission
Posted by Michael Ho <kw...@cloudera.com>.
And to provide more context for Impala's query cancellation, the
cancellation is asynchronous. In other words, the coordinator is only
responsible for notifying the remote Impalad to shut down the fragment
instances at a convenient time. The coordinator will not wait for
acknowledgement from all Impalad before proceeding. The assumption is that
the executor threads will periodically poll for cancellation request and
free up the resources soon after it sees one.
On Fri, Oct 13, 2017 at 1:10 PM, Michael Ho <kw...@cloudera.com> wrote:
> The coordinator RPC to cancel the fragment instance usually has much
> smaller payload compared to the RPC for transmitting data between Impalad
> nodes so it has a higher chance of coming through. I agree that, when the
> network is in bad shape, cancellation requests may not reach all Impalad
> nodes. However, keep in mind that it's not necessarily the case that the
> entire network is swamped. It can be that one or a few nodes in the cluster
> which were pounded on with a lot of network traffic. In those cases, it
> seems all other fragment instances on other Impalad nodes will still block
> waiting for these few overloaded during cancellation and hold up memory
> resources on those Impalad nodes.
>
> Thanks,
> Michael
>
> On Fri, Oct 13, 2017 at 11:30 AM, Dan Burkert <da...@cloudera.com> wrote:
>
>> I'm still hung up on the premise that if the network is down,
>> cancellation should still be expected to complete in bounded time. If the
>> network is partitioned or congested to the point of non-delivery, how is
>> the coordinator going to deliver the cancellation messages to the fragments
>> in the first place? If we're allowing for arbitrary network delays, there
>> is no way to make this guarantee. We're laser focussed on this issue of
>> cancelling mid-transmission RPCs, but transmission is a very minor slice of
>> the whole RPC lifecycle, and I don't believe it's worth complicating the
>> RPC protocol or lifecycle over.
>>
>> - Dan
>>
>> On Fri, Oct 13, 2017 at 11:01 AM, Sailesh Mukil <sa...@cloudera.com>
>> wrote:
>>
>>> On Fri, Oct 13, 2017 at 9:37 AM, Michael Ho <kw...@cloudera.com> wrote:
>>>
>>> > There were some recent discussions about the proposed mechanism to
>>> cancel
>>> > an RPC in mid-transmission. I am writing this email to summarize the
>>> > problem cancellation is trying to solve and some proposed solutions.
>>> >
>>> > As part of IMPALA-2567 <https://issues.apache.org/jir
>>> a/browse/IMPALA-2567>,
>>>
>>> > Impala is porting its RPC facility to utilize the Kudu RPC library. One
>>> > major feature of KRPC is asynchronous RPC. This prevents the need for
>>> > Impala to create a separate thread for each communication channel
>>> between
>>> > fragment instances running on different Impala backends. When an
>>> > asynchronous RPC is in progress, the KRPC code needs to hold on to the
>>> RPC
>>> > payload provided and owned by a fragment instance. We have a singleton
>>> > messenger instance per Impalad node. The connection between two hosts
>>> is
>>> > shared by different fragment instances.
>>> >
>>> > At any point when the RPC is in-progress, the coordinator can cancel
>>> all
>>> > fragment instances of a query for various reasons. In which case, the
>>> > fragment execution threads need to cancel their in-flight RPC and free
>>> up
>>> > the RPC payloads (e.g. which are passed as sidecars of the RPC request)
>>> > within a reasonable amount of time (e.g. 500ms). Currently in the KRPC
>>> > code, when a cancellation is requested when an outbound call is in
>>> SENDING
>>> > state (i.e. it has potentially sent some or the whole header to the
>>> remote
>>> > server), it will wait for the entire payload to be sent before
>>> invoking the
>>> > completion callback specified in the async RPC request. At which
>>> point, the
>>> > KRPC client can assume that the RPC payload is no longer referenced by
>>> the
>>> > KRPC code and it can be freed.
>>> >
>>> > When the network connections between Impalad backends is functioning
>>> well,
>>> > it shouldn't be long before the payload is done sending as Impalad
>>> > internally tries to bound the payload size to 8MB (but not necessarily
>>> > guaranteed in some degenerated case e.g. row with 1GB string). However,
>>> > when the network gets congested or is very lossy, there is in theory no
>>> > bound on how long it takes to finish sending the payload. In other
>>> words,
>>> > the cancellation of the RPC can take unbounded amount of time to
>>> complete
>>> > and thus the cancellation of a fragment instance can also take
>>> infinitely
>>> > long. This is a chicken-and-egg problem. Usually, the user cancels the
>>> > query because the poor network performance caused it to run for too
>>> long.
>>> > With the unbounded wait, the user cannot even cancel the query within a
>>> > bounded period of time.
>>> >
>>> > Similar problem arises when an RPC times out. As the KRPC code stands
>>> > today, there is no guarantee that the payload is not referenced by the
>>> KRPC
>>> > code when the completion callback is invoked due to timeout. The client
>>> > doesn't get any notification on when the payload can be freed. In
>>> theory,
>>> > this can be solved by deferring the callback until the payload has been
>>> > completely sent. However, this may not always be desirable due to the
>>> > unbounded wait mentioned above. For example, if a client requests a
>>> timeout
>>> > of 60s, it's not desirable to invoke the callback say 300s after the
>>> > timeout.
>>> >
>>> > If you agree that the above are indeed problems, we need a RPC
>>> > cancellation mechanism which meets the following requirements:
>>> >
>>> > - the RPC payload is no longer referenced by the KRPC code when the
>>> > completion callback is invoked.
>>> > - the time between the cancellation request and the invocation of the
>>> > completion callback should be reasonably bounded.
>>> >
>>> > The followings are two different proposals floated:
>>> >
>>> > 1. Upgrade the KRPC protocol to insert a footer at the very end of each
>>> > outbound call transfer (i.e. after header, serialized protobuf request,
>>> > sidecars etc). The footer contains a flag which is parsed by the
>>> server to
>>> > determine whether the inbound call should be discarded. Normally, this
>>> flag
>>> > will be cleared. When an outbound call is cancelled while it's still
>>> in the
>>> > SENDING state, the footer is modified in place to have the flag set. In
>>> > addition, the sidecars of the outbound call transfers are modified to
>>> point
>>> > to a dummy buffer. In this way, the slices in an outbound transfer no
>>> > longer references the original payload of the RPC and the footer
>>> guarantees
>>> > that whatever is received by the server will be discarded so it's okay
>>> to
>>> > replace the original payload with dummy bytes. This allows the
>>> completion
>>> > callback to be invoked without waiting for the entire payload to be
>>> sent.
>>> >
>>> > Pros:
>>> > - relatively low overhead in network consumption (1 byte per outbound
>>> call
>>> > transfer)
>>> > - cancellation is complete once the reactor thread handles the
>>> > cancellation event and is not reliant on the network condition.
>>> >
>>> > Cons:
>>> > - the remaining bytes of the payload still need to be sent across a
>>> > congested network after cancellation.
>>> > - added complexity to the protocol.
>>> >
>>> > The details of the proposal is in KUDU-2065
>>> > <https://issues.apache.org/jira/browse/KUDU-2065>. The patch which
>>> > implemented this idea is at https://gerrit.cloudera.org/#/c/7599/
>>> > It was later reverted due to KUDU-2110
>>> > <https://issues.apache.org/jira/browse/KUDU-2110>.
>>>
>>> >
>>> > 2. Keep the KRPC protocol unchanged. When an outbound call is cancelled
>>> > when it's in SENDING state, register a timer with a timeout (e.g.
>>> 500ms).
>>> > When the timer expires, invoke a callback to call shutdown(fd,
>>> SHUT_WR) on
>>> > the socket. This shuts down one-half of the connection and prevents
>>> more
>>> > bytes from being sent while allowing bytes to be read on the client
>>> side.
>>> > recv() will eventually return 0 on the server side. After all other
>>> > in-flight RPCs for that connection have been replied to, the server
>>> should
>>> > close the connection. The client, after reading all the replies, should
>>> > also destroy the Connection object. All RPCs behind the cancelled the
>>> RPC
>>> > in the outbound queue on the client side are aborted and need to be
>>> retried.
>>> >
>>> > Pro:
>>> > - stop piling on a congested network after cancelling an RPC.
>>> > - cancellation is still bounded in time.
>>> >
>>> > Con:
>>> > - For Impalad, each host may be connected to all other hosts in the
>>> > cluster. In the worst case, this means closing all n connections. In
>>> total,
>>> > that could be n^2 connection being re-established concurrently. This
>>> can
>>> > easily trigger negotiation timeout in a secure cluster based on our
>>> testing
>>> > on bolt1k cluster, leading to unpredictable failures.
>>> >
>>> > - The RPCs behind the cancelled RPC in the outbound queue cannot be
>>> > retried until all previous in-flight RPCs have been replied to. This
>>> can be
>>> > unboundedly long if the remote server is overloaded.
>>> >
>>> >
>>> Regarding option 2, although this may work well with certain workloads,
>>> I'm
>>> not sure that this would be the right way to go about it. Also, in some
>>> scenarios, it could be very slow.
>>>
>>> From the viewpoint of Impala, cancellation can happen in many scenarios.
>>> Other than a query being cancelled by a user, cancellations also happen
>>> when there are runtime errors (OOM, found a bad file in the storage
>>> layer,
>>> etc.)
>>>
>>> Even aside from the negotiation storm mentioned above by Michael, if we
>>> were to tear down the connection for every cancel request we get, it will
>>> cause a slow down for all other queries running on those nodes as well,
>>> since they have to now wait longer until the connection is re-setup.
>>>
>>> We can imagine a scenario where the cluster is under high stress leading
>>> quite a few queries to OOM, which would send out cancellations to all in
>>> flight RPCs for those respective queries, causing a chain of
>>> cancellations.
>>> We can also imagine multiple CANCEL requests from different queries being
>>> queued for the same connection, so the first CANCEL request tears down
>>> the
>>> connection, and sets it up again, only to be torn down again by another
>>> queued CANCEL request. Sure, we can try to detect multiple CANCEL
>>> requests
>>> in a queue and do it only once, but that goes back to adding complexity
>>> to
>>> the protocol.
>>>
>>> To my knowledge, I don't know of a RPC library that takes option 2, but I
>>> know at least one that does something similar to option 1 (grpc). That's
>>> my
>>> two cents on the issue, but I may have missed out thinking about it from
>>> different perspectives, so feel free to correct me if I got something
>>> wrong.
>>>
>>> Please feel free to comment on the proposals above or suggest new ideas.
>>> >
>>> > --
>>> > Thanks,
>>> > Michael
>>> >
>>>
>>
>>
>
>
> --
> Thanks,
> Michael
>
--
Thanks,
Michael
Re: RFC: Canceling RPC in mid-transmission
Posted by Michael Ho <kw...@cloudera.com>.
The coordinator RPC to cancel the fragment instance usually has much
smaller payload compared to the RPC for transmitting data between Impalad
nodes so it has a higher chance of coming through. I agree that, when the
network is in bad shape, cancellation requests may not reach all Impalad
nodes. However, keep in mind that it's not necessarily the case that the
entire network is swamped. It can be that one or a few nodes in the cluster
which were pounded on with a lot of network traffic. In those cases, it
seems all other fragment instances on other Impalad nodes will still block
waiting for these few overloaded during cancellation and hold up memory
resources on those Impalad nodes.
Thanks,
Michael
On Fri, Oct 13, 2017 at 11:30 AM, Dan Burkert <da...@cloudera.com> wrote:
> I'm still hung up on the premise that if the network is down, cancellation
> should still be expected to complete in bounded time. If the network is
> partitioned or congested to the point of non-delivery, how is the
> coordinator going to deliver the cancellation messages to the fragments in
> the first place? If we're allowing for arbitrary network delays, there is
> no way to make this guarantee. We're laser focussed on this issue of
> cancelling mid-transmission RPCs, but transmission is a very minor slice of
> the whole RPC lifecycle, and I don't believe it's worth complicating the
> RPC protocol or lifecycle over.
>
> - Dan
>
> On Fri, Oct 13, 2017 at 11:01 AM, Sailesh Mukil <sa...@cloudera.com>
> wrote:
>
>> On Fri, Oct 13, 2017 at 9:37 AM, Michael Ho <kw...@cloudera.com> wrote:
>>
>> > There were some recent discussions about the proposed mechanism to
>> cancel
>> > an RPC in mid-transmission. I am writing this email to summarize the
>> > problem cancellation is trying to solve and some proposed solutions.
>> >
>> > As part of IMPALA-2567 <https://issues.apache.org/jir
>> a/browse/IMPALA-2567>,
>>
>> > Impala is porting its RPC facility to utilize the Kudu RPC library. One
>> > major feature of KRPC is asynchronous RPC. This prevents the need for
>> > Impala to create a separate thread for each communication channel
>> between
>> > fragment instances running on different Impala backends. When an
>> > asynchronous RPC is in progress, the KRPC code needs to hold on to the
>> RPC
>> > payload provided and owned by a fragment instance. We have a singleton
>> > messenger instance per Impalad node. The connection between two hosts is
>> > shared by different fragment instances.
>> >
>> > At any point when the RPC is in-progress, the coordinator can cancel all
>> > fragment instances of a query for various reasons. In which case, the
>> > fragment execution threads need to cancel their in-flight RPC and free
>> up
>> > the RPC payloads (e.g. which are passed as sidecars of the RPC request)
>> > within a reasonable amount of time (e.g. 500ms). Currently in the KRPC
>> > code, when a cancellation is requested when an outbound call is in
>> SENDING
>> > state (i.e. it has potentially sent some or the whole header to the
>> remote
>> > server), it will wait for the entire payload to be sent before invoking
>> the
>> > completion callback specified in the async RPC request. At which point,
>> the
>> > KRPC client can assume that the RPC payload is no longer referenced by
>> the
>> > KRPC code and it can be freed.
>> >
>> > When the network connections between Impalad backends is functioning
>> well,
>> > it shouldn't be long before the payload is done sending as Impalad
>> > internally tries to bound the payload size to 8MB (but not necessarily
>> > guaranteed in some degenerated case e.g. row with 1GB string). However,
>> > when the network gets congested or is very lossy, there is in theory no
>> > bound on how long it takes to finish sending the payload. In other
>> words,
>> > the cancellation of the RPC can take unbounded amount of time to
>> complete
>> > and thus the cancellation of a fragment instance can also take
>> infinitely
>> > long. This is a chicken-and-egg problem. Usually, the user cancels the
>> > query because the poor network performance caused it to run for too
>> long.
>> > With the unbounded wait, the user cannot even cancel the query within a
>> > bounded period of time.
>> >
>> > Similar problem arises when an RPC times out. As the KRPC code stands
>> > today, there is no guarantee that the payload is not referenced by the
>> KRPC
>> > code when the completion callback is invoked due to timeout. The client
>> > doesn't get any notification on when the payload can be freed. In
>> theory,
>> > this can be solved by deferring the callback until the payload has been
>> > completely sent. However, this may not always be desirable due to the
>> > unbounded wait mentioned above. For example, if a client requests a
>> timeout
>> > of 60s, it's not desirable to invoke the callback say 300s after the
>> > timeout.
>> >
>> > If you agree that the above are indeed problems, we need a RPC
>> > cancellation mechanism which meets the following requirements:
>> >
>> > - the RPC payload is no longer referenced by the KRPC code when the
>> > completion callback is invoked.
>> > - the time between the cancellation request and the invocation of the
>> > completion callback should be reasonably bounded.
>> >
>> > The followings are two different proposals floated:
>> >
>> > 1. Upgrade the KRPC protocol to insert a footer at the very end of each
>> > outbound call transfer (i.e. after header, serialized protobuf request,
>> > sidecars etc). The footer contains a flag which is parsed by the server
>> to
>> > determine whether the inbound call should be discarded. Normally, this
>> flag
>> > will be cleared. When an outbound call is cancelled while it's still in
>> the
>> > SENDING state, the footer is modified in place to have the flag set. In
>> > addition, the sidecars of the outbound call transfers are modified to
>> point
>> > to a dummy buffer. In this way, the slices in an outbound transfer no
>> > longer references the original payload of the RPC and the footer
>> guarantees
>> > that whatever is received by the server will be discarded so it's okay
>> to
>> > replace the original payload with dummy bytes. This allows the
>> completion
>> > callback to be invoked without waiting for the entire payload to be
>> sent.
>> >
>> > Pros:
>> > - relatively low overhead in network consumption (1 byte per outbound
>> call
>> > transfer)
>> > - cancellation is complete once the reactor thread handles the
>> > cancellation event and is not reliant on the network condition.
>> >
>> > Cons:
>> > - the remaining bytes of the payload still need to be sent across a
>> > congested network after cancellation.
>> > - added complexity to the protocol.
>> >
>> > The details of the proposal is in KUDU-2065
>> > <https://issues.apache.org/jira/browse/KUDU-2065>. The patch which
>> > implemented this idea is at https://gerrit.cloudera.org/#/c/7599/
>> > It was later reverted due to KUDU-2110
>> > <https://issues.apache.org/jira/browse/KUDU-2110>.
>>
>> >
>> > 2. Keep the KRPC protocol unchanged. When an outbound call is cancelled
>> > when it's in SENDING state, register a timer with a timeout (e.g.
>> 500ms).
>> > When the timer expires, invoke a callback to call shutdown(fd, SHUT_WR)
>> on
>> > the socket. This shuts down one-half of the connection and prevents more
>> > bytes from being sent while allowing bytes to be read on the client
>> side.
>> > recv() will eventually return 0 on the server side. After all other
>> > in-flight RPCs for that connection have been replied to, the server
>> should
>> > close the connection. The client, after reading all the replies, should
>> > also destroy the Connection object. All RPCs behind the cancelled the
>> RPC
>> > in the outbound queue on the client side are aborted and need to be
>> retried.
>> >
>> > Pro:
>> > - stop piling on a congested network after cancelling an RPC.
>> > - cancellation is still bounded in time.
>> >
>> > Con:
>> > - For Impalad, each host may be connected to all other hosts in the
>> > cluster. In the worst case, this means closing all n connections. In
>> total,
>> > that could be n^2 connection being re-established concurrently. This can
>> > easily trigger negotiation timeout in a secure cluster based on our
>> testing
>> > on bolt1k cluster, leading to unpredictable failures.
>> >
>> > - The RPCs behind the cancelled RPC in the outbound queue cannot be
>> > retried until all previous in-flight RPCs have been replied to. This
>> can be
>> > unboundedly long if the remote server is overloaded.
>> >
>> >
>> Regarding option 2, although this may work well with certain workloads,
>> I'm
>> not sure that this would be the right way to go about it. Also, in some
>> scenarios, it could be very slow.
>>
>> From the viewpoint of Impala, cancellation can happen in many scenarios.
>> Other than a query being cancelled by a user, cancellations also happen
>> when there are runtime errors (OOM, found a bad file in the storage layer,
>> etc.)
>>
>> Even aside from the negotiation storm mentioned above by Michael, if we
>> were to tear down the connection for every cancel request we get, it will
>> cause a slow down for all other queries running on those nodes as well,
>> since they have to now wait longer until the connection is re-setup.
>>
>> We can imagine a scenario where the cluster is under high stress leading
>> quite a few queries to OOM, which would send out cancellations to all in
>> flight RPCs for those respective queries, causing a chain of
>> cancellations.
>> We can also imagine multiple CANCEL requests from different queries being
>> queued for the same connection, so the first CANCEL request tears down the
>> connection, and sets it up again, only to be torn down again by another
>> queued CANCEL request. Sure, we can try to detect multiple CANCEL requests
>> in a queue and do it only once, but that goes back to adding complexity to
>> the protocol.
>>
>> To my knowledge, I don't know of a RPC library that takes option 2, but I
>> know at least one that does something similar to option 1 (grpc). That's
>> my
>> two cents on the issue, but I may have missed out thinking about it from
>> different perspectives, so feel free to correct me if I got something
>> wrong.
>>
>> Please feel free to comment on the proposals above or suggest new ideas.
>> >
>> > --
>> > Thanks,
>> > Michael
>> >
>>
>
>
--
Thanks,
Michael
Re: RFC: Canceling RPC in mid-transmission
Posted by Dan Burkert <da...@cloudera.com>.
I'm still hung up on the premise that if the network is down, cancellation
should still be expected to complete in bounded time. If the network is
partitioned or congested to the point of non-delivery, how is the
coordinator going to deliver the cancellation messages to the fragments in
the first place? If we're allowing for arbitrary network delays, there is
no way to make this guarantee. We're laser focussed on this issue of
cancelling mid-transmission RPCs, but transmission is a very minor slice of
the whole RPC lifecycle, and I don't believe it's worth complicating the
RPC protocol or lifecycle over.
- Dan
On Fri, Oct 13, 2017 at 11:01 AM, Sailesh Mukil <sa...@cloudera.com>
wrote:
> On Fri, Oct 13, 2017 at 9:37 AM, Michael Ho <kw...@cloudera.com> wrote:
>
> > There were some recent discussions about the proposed mechanism to cancel
> > an RPC in mid-transmission. I am writing this email to summarize the
> > problem cancellation is trying to solve and some proposed solutions.
> >
> > As part of IMPALA-2567 <https://issues.apache.org/
> jira/browse/IMPALA-2567>,
> > Impala is porting its RPC facility to utilize the Kudu RPC library. One
> > major feature of KRPC is asynchronous RPC. This prevents the need for
> > Impala to create a separate thread for each communication channel between
> > fragment instances running on different Impala backends. When an
> > asynchronous RPC is in progress, the KRPC code needs to hold on to the
> RPC
> > payload provided and owned by a fragment instance. We have a singleton
> > messenger instance per Impalad node. The connection between two hosts is
> > shared by different fragment instances.
> >
> > At any point when the RPC is in-progress, the coordinator can cancel all
> > fragment instances of a query for various reasons. In which case, the
> > fragment execution threads need to cancel their in-flight RPC and free up
> > the RPC payloads (e.g. which are passed as sidecars of the RPC request)
> > within a reasonable amount of time (e.g. 500ms). Currently in the KRPC
> > code, when a cancellation is requested when an outbound call is in
> SENDING
> > state (i.e. it has potentially sent some or the whole header to the
> remote
> > server), it will wait for the entire payload to be sent before invoking
> the
> > completion callback specified in the async RPC request. At which point,
> the
> > KRPC client can assume that the RPC payload is no longer referenced by
> the
> > KRPC code and it can be freed.
> >
> > When the network connections between Impalad backends is functioning
> well,
> > it shouldn't be long before the payload is done sending as Impalad
> > internally tries to bound the payload size to 8MB (but not necessarily
> > guaranteed in some degenerated case e.g. row with 1GB string). However,
> > when the network gets congested or is very lossy, there is in theory no
> > bound on how long it takes to finish sending the payload. In other words,
> > the cancellation of the RPC can take unbounded amount of time to complete
> > and thus the cancellation of a fragment instance can also take infinitely
> > long. This is a chicken-and-egg problem. Usually, the user cancels the
> > query because the poor network performance caused it to run for too long.
> > With the unbounded wait, the user cannot even cancel the query within a
> > bounded period of time.
> >
> > Similar problem arises when an RPC times out. As the KRPC code stands
> > today, there is no guarantee that the payload is not referenced by the
> KRPC
> > code when the completion callback is invoked due to timeout. The client
> > doesn't get any notification on when the payload can be freed. In theory,
> > this can be solved by deferring the callback until the payload has been
> > completely sent. However, this may not always be desirable due to the
> > unbounded wait mentioned above. For example, if a client requests a
> timeout
> > of 60s, it's not desirable to invoke the callback say 300s after the
> > timeout.
> >
> > If you agree that the above are indeed problems, we need a RPC
> > cancellation mechanism which meets the following requirements:
> >
> > - the RPC payload is no longer referenced by the KRPC code when the
> > completion callback is invoked.
> > - the time between the cancellation request and the invocation of the
> > completion callback should be reasonably bounded.
> >
> > The followings are two different proposals floated:
> >
> > 1. Upgrade the KRPC protocol to insert a footer at the very end of each
> > outbound call transfer (i.e. after header, serialized protobuf request,
> > sidecars etc). The footer contains a flag which is parsed by the server
> to
> > determine whether the inbound call should be discarded. Normally, this
> flag
> > will be cleared. When an outbound call is cancelled while it's still in
> the
> > SENDING state, the footer is modified in place to have the flag set. In
> > addition, the sidecars of the outbound call transfers are modified to
> point
> > to a dummy buffer. In this way, the slices in an outbound transfer no
> > longer references the original payload of the RPC and the footer
> guarantees
> > that whatever is received by the server will be discarded so it's okay to
> > replace the original payload with dummy bytes. This allows the completion
> > callback to be invoked without waiting for the entire payload to be sent.
> >
> > Pros:
> > - relatively low overhead in network consumption (1 byte per outbound
> call
> > transfer)
> > - cancellation is complete once the reactor thread handles the
> > cancellation event and is not reliant on the network condition.
> >
> > Cons:
> > - the remaining bytes of the payload still need to be sent across a
> > congested network after cancellation.
> > - added complexity to the protocol.
> >
> > The details of the proposal is in KUDU-2065
> > <https://issues.apache.org/jira/browse/KUDU-2065>. The patch which
> > implemented this idea is at https://gerrit.cloudera.org/#/c/7599/
> > It was later reverted due to KUDU-2110
> > <https://issues.apache.org/jira/browse/KUDU-2110>.
> >
> > 2. Keep the KRPC protocol unchanged. When an outbound call is cancelled
> > when it's in SENDING state, register a timer with a timeout (e.g. 500ms).
> > When the timer expires, invoke a callback to call shutdown(fd, SHUT_WR)
> on
> > the socket. This shuts down one-half of the connection and prevents more
> > bytes from being sent while allowing bytes to be read on the client side.
> > recv() will eventually return 0 on the server side. After all other
> > in-flight RPCs for that connection have been replied to, the server
> should
> > close the connection. The client, after reading all the replies, should
> > also destroy the Connection object. All RPCs behind the cancelled the RPC
> > in the outbound queue on the client side are aborted and need to be
> retried.
> >
> > Pro:
> > - stop piling on a congested network after cancelling an RPC.
> > - cancellation is still bounded in time.
> >
> > Con:
> > - For Impalad, each host may be connected to all other hosts in the
> > cluster. In the worst case, this means closing all n connections. In
> total,
> > that could be n^2 connection being re-established concurrently. This can
> > easily trigger negotiation timeout in a secure cluster based on our
> testing
> > on bolt1k cluster, leading to unpredictable failures.
> >
> > - The RPCs behind the cancelled RPC in the outbound queue cannot be
> > retried until all previous in-flight RPCs have been replied to. This can
> be
> > unboundedly long if the remote server is overloaded.
> >
> >
> Regarding option 2, although this may work well with certain workloads, I'm
> not sure that this would be the right way to go about it. Also, in some
> scenarios, it could be very slow.
>
> From the viewpoint of Impala, cancellation can happen in many scenarios.
> Other than a query being cancelled by a user, cancellations also happen
> when there are runtime errors (OOM, found a bad file in the storage layer,
> etc.)
>
> Even aside from the negotiation storm mentioned above by Michael, if we
> were to tear down the connection for every cancel request we get, it will
> cause a slow down for all other queries running on those nodes as well,
> since they have to now wait longer until the connection is re-setup.
>
> We can imagine a scenario where the cluster is under high stress leading
> quite a few queries to OOM, which would send out cancellations to all in
> flight RPCs for those respective queries, causing a chain of cancellations.
> We can also imagine multiple CANCEL requests from different queries being
> queued for the same connection, so the first CANCEL request tears down the
> connection, and sets it up again, only to be torn down again by another
> queued CANCEL request. Sure, we can try to detect multiple CANCEL requests
> in a queue and do it only once, but that goes back to adding complexity to
> the protocol.
>
> To my knowledge, I don't know of a RPC library that takes option 2, but I
> know at least one that does something similar to option 1 (grpc). That's my
> two cents on the issue, but I may have missed out thinking about it from
> different perspectives, so feel free to correct me if I got something
> wrong.
>
> Please feel free to comment on the proposals above or suggest new ideas.
> >
> > --
> > Thanks,
> > Michael
> >
>
Re: RFC: Canceling RPC in mid-transmission
Posted by Sailesh Mukil <sa...@cloudera.com>.
On Fri, Oct 13, 2017 at 9:37 AM, Michael Ho <kw...@cloudera.com> wrote:
> There were some recent discussions about the proposed mechanism to cancel
> an RPC in mid-transmission. I am writing this email to summarize the
> problem cancellation is trying to solve and some proposed solutions.
>
> As part of IMPALA-2567 <https://issues.apache.org/jira/browse/IMPALA-2567>,
> Impala is porting its RPC facility to utilize the Kudu RPC library. One
> major feature of KRPC is asynchronous RPC. This prevents the need for
> Impala to create a separate thread for each communication channel between
> fragment instances running on different Impala backends. When an
> asynchronous RPC is in progress, the KRPC code needs to hold on to the RPC
> payload provided and owned by a fragment instance. We have a singleton
> messenger instance per Impalad node. The connection between two hosts is
> shared by different fragment instances.
>
> At any point when the RPC is in-progress, the coordinator can cancel all
> fragment instances of a query for various reasons. In which case, the
> fragment execution threads need to cancel their in-flight RPC and free up
> the RPC payloads (e.g. which are passed as sidecars of the RPC request)
> within a reasonable amount of time (e.g. 500ms). Currently in the KRPC
> code, when a cancellation is requested when an outbound call is in SENDING
> state (i.e. it has potentially sent some or the whole header to the remote
> server), it will wait for the entire payload to be sent before invoking the
> completion callback specified in the async RPC request. At which point, the
> KRPC client can assume that the RPC payload is no longer referenced by the
> KRPC code and it can be freed.
>
> When the network connections between Impalad backends is functioning well,
> it shouldn't be long before the payload is done sending as Impalad
> internally tries to bound the payload size to 8MB (but not necessarily
> guaranteed in some degenerated case e.g. row with 1GB string). However,
> when the network gets congested or is very lossy, there is in theory no
> bound on how long it takes to finish sending the payload. In other words,
> the cancellation of the RPC can take unbounded amount of time to complete
> and thus the cancellation of a fragment instance can also take infinitely
> long. This is a chicken-and-egg problem. Usually, the user cancels the
> query because the poor network performance caused it to run for too long.
> With the unbounded wait, the user cannot even cancel the query within a
> bounded period of time.
>
> Similar problem arises when an RPC times out. As the KRPC code stands
> today, there is no guarantee that the payload is not referenced by the KRPC
> code when the completion callback is invoked due to timeout. The client
> doesn't get any notification on when the payload can be freed. In theory,
> this can be solved by deferring the callback until the payload has been
> completely sent. However, this may not always be desirable due to the
> unbounded wait mentioned above. For example, if a client requests a timeout
> of 60s, it's not desirable to invoke the callback say 300s after the
> timeout.
>
> If you agree that the above are indeed problems, we need a RPC
> cancellation mechanism which meets the following requirements:
>
> - the RPC payload is no longer referenced by the KRPC code when the
> completion callback is invoked.
> - the time between the cancellation request and the invocation of the
> completion callback should be reasonably bounded.
>
> The followings are two different proposals floated:
>
> 1. Upgrade the KRPC protocol to insert a footer at the very end of each
> outbound call transfer (i.e. after header, serialized protobuf request,
> sidecars etc). The footer contains a flag which is parsed by the server to
> determine whether the inbound call should be discarded. Normally, this flag
> will be cleared. When an outbound call is cancelled while it's still in the
> SENDING state, the footer is modified in place to have the flag set. In
> addition, the sidecars of the outbound call transfers are modified to point
> to a dummy buffer. In this way, the slices in an outbound transfer no
> longer references the original payload of the RPC and the footer guarantees
> that whatever is received by the server will be discarded so it's okay to
> replace the original payload with dummy bytes. This allows the completion
> callback to be invoked without waiting for the entire payload to be sent.
>
> Pros:
> - relatively low overhead in network consumption (1 byte per outbound call
> transfer)
> - cancellation is complete once the reactor thread handles the
> cancellation event and is not reliant on the network condition.
>
> Cons:
> - the remaining bytes of the payload still need to be sent across a
> congested network after cancellation.
> - added complexity to the protocol.
>
> The details of the proposal is in KUDU-2065
> <https://issues.apache.org/jira/browse/KUDU-2065>. The patch which
> implemented this idea is at https://gerrit.cloudera.org/#/c/7599/
> It was later reverted due to KUDU-2110
> <https://issues.apache.org/jira/browse/KUDU-2110>.
>
> 2. Keep the KRPC protocol unchanged. When an outbound call is cancelled
> when it's in SENDING state, register a timer with a timeout (e.g. 500ms).
> When the timer expires, invoke a callback to call shutdown(fd, SHUT_WR) on
> the socket. This shuts down one-half of the connection and prevents more
> bytes from being sent while allowing bytes to be read on the client side.
> recv() will eventually return 0 on the server side. After all other
> in-flight RPCs for that connection have been replied to, the server should
> close the connection. The client, after reading all the replies, should
> also destroy the Connection object. All RPCs behind the cancelled the RPC
> in the outbound queue on the client side are aborted and need to be retried.
>
> Pro:
> - stop piling on a congested network after cancelling an RPC.
> - cancellation is still bounded in time.
>
> Con:
> - For Impalad, each host may be connected to all other hosts in the
> cluster. In the worst case, this means closing all n connections. In total,
> that could be n^2 connection being re-established concurrently. This can
> easily trigger negotiation timeout in a secure cluster based on our testing
> on bolt1k cluster, leading to unpredictable failures.
>
> - The RPCs behind the cancelled RPC in the outbound queue cannot be
> retried until all previous in-flight RPCs have been replied to. This can be
> unboundedly long if the remote server is overloaded.
>
>
Regarding option 2, although this may work well with certain workloads, I'm
not sure that this would be the right way to go about it. Also, in some
scenarios, it could be very slow.
From the viewpoint of Impala, cancellation can happen in many scenarios.
Other than a query being cancelled by a user, cancellations also happen
when there are runtime errors (OOM, found a bad file in the storage layer,
etc.)
Even aside from the negotiation storm mentioned above by Michael, if we
were to tear down the connection for every cancel request we get, it will
cause a slow down for all other queries running on those nodes as well,
since they have to now wait longer until the connection is re-setup.
We can imagine a scenario where the cluster is under high stress leading
quite a few queries to OOM, which would send out cancellations to all in
flight RPCs for those respective queries, causing a chain of cancellations.
We can also imagine multiple CANCEL requests from different queries being
queued for the same connection, so the first CANCEL request tears down the
connection, and sets it up again, only to be torn down again by another
queued CANCEL request. Sure, we can try to detect multiple CANCEL requests
in a queue and do it only once, but that goes back to adding complexity to
the protocol.
To my knowledge, I don't know of a RPC library that takes option 2, but I
know at least one that does something similar to option 1 (grpc). That's my
two cents on the issue, but I may have missed out thinking about it from
different perspectives, so feel free to correct me if I got something
wrong.
Please feel free to comment on the proposals above or suggest new ideas.
>
> --
> Thanks,
> Michael
>