summaryrefslogtreecommitdiff
path: root/src/mongo/s/async_requests_sender.h
blob: daf48c558aad2d73d7061de957b3abafc92140b9 (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
/**
 *    Copyright (C) 2017 MongoDB Inc.
 *
 *    This program is free software: you can redistribute it and/or  modify
 *    it under the terms of the GNU Affero General Public License, version 3,
 *    as published by the Free Software Foundation.
 *
 *    This program is distributed in the hope that it will be useful,
 *    but WITHOUT ANY WARRANTY; without even the implied warranty of
 *    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 *    GNU Affero General Public License for more details.
 *
 *    You should have received a copy of the GNU Affero General Public License
 *    along with this program.  If not, see <http://www.gnu.org/licenses/>.
 *
 *    As a special exception, the copyright holders give permission to link the
 *    code of portions of this program with the OpenSSL library under certain
 *    conditions as described in each individual source file and distribute
 *    linked combinations including the program with the OpenSSL library. You
 *    must comply with the GNU Affero General Public License in all respects for
 *    all of the code used other than as permitted herein. If you modify file(s)
 *    with this exception, you may extend this exception to your version of the
 *    file(s), but you are not obligated to do so. If you do not wish to do so,
 *    delete this exception statement from your version. If you delete this
 *    exception statement from all source files in the program, then also delete
 *    it in the license file.
 */

#pragma once

#include <boost/optional.hpp>
#include <vector>

#include "mongo/base/disallow_copying.h"
#include "mongo/base/status_with.h"
#include "mongo/bson/bsonobj.h"
#include "mongo/client/read_preference.h"
#include "mongo/executor/remote_command_response.h"
#include "mongo/executor/task_executor.h"
#include "mongo/s/client/shard.h"
#include "mongo/s/shard_id.h"
#include "mongo/stdx/mutex.h"
#include "mongo/util/concurrency/notification.h"
#include "mongo/util/net/hostandport.h"
#include "mongo/util/time_support.h"

namespace mongo {

/**
 * The AsyncRequestsSender allows for sending requests to a set of remote shards in parallel and
 * automatically retrying on retriable errors according to a RetryPolicy. It can also allow for
 * retrieving partial results by ignoring shards that return errors.
 *
 * Work on remote nodes is accomplished by scheduling remote work in a TaskExecutor's event loop.
 *
 * Typical usage is:
 *
 * AsyncRequestsSender ars(txn, executor, db, requests, readPrefSetting);  // schedule the requests
 * auto responses = ars.waitForResponses(txn);  // wait for responses; retries on retriable erors
 *
 * Additionally, you can interrupt() (if you want waitForResponses() to wait for responses for
 * outstanding requests but stop scheduling retries) or kill() (if you want to cancel outstanding
 * requests) the ARS from another thread.
 *
 * Does not throw exceptions.
 */
class AsyncRequestsSender {
    MONGO_DISALLOW_COPYING(AsyncRequestsSender);

public:
    /**
     * Defines a request to a remote shard that can be run by the ARS.
     */
    struct Request {
        Request(ShardId shardId, BSONObj cmdObj);

        // ShardId of the shard to which the command will be sent.
        const ShardId shardId;

        // The command object to send to the remote host.
        const BSONObj cmdObj;
    };

    /**
     * Defines a response for a request to a remote shard.
     */
    struct Response {
        // Constructor for a response that was successfully received.
        Response(executor::RemoteCommandResponse response, HostAndPort hp);

        // Constructor that specifies the reason the response was not successfully received.
        Response(Status status, boost::optional<HostAndPort> hp);

        // The response or error from the remote.
        StatusWith<executor::RemoteCommandResponse> swResponse;

        // The exact host on which the remote command was run. Is unset if the shard could not be
        // found or no shard hosts matching the readPreference could be found.
        boost::optional<HostAndPort> shardHostAndPort;
    };

    /**
     * Constructs a new AsyncRequestsSender. The TaskExecutor* must remain valid for the lifetime of
     * the ARS.
     */
    AsyncRequestsSender(OperationContext* txn,
                        executor::TaskExecutor* executor,
                        StringData db,
                        const std::vector<AsyncRequestsSender::Request>& requests,
                        const ReadPreferenceSetting& readPreference,
                        bool allowPartialResults = false);

    ~AsyncRequestsSender();

    /**
     * Returns a vector containing the responses or errors for each remote in the same order as the
     * input vector that was passed in the constructor.
     *
     * If we were killed, returns immediately.
     * If we were interrupted, returns when any outstanding requests have completed.
     * Otherwise, returns when each remote has received a response or error.
     *
     * Must only be called once.
     */
    std::vector<Response> waitForResponses(OperationContext* txn);

    /**
     * Stops the ARS from retrying requests. Causes waitForResponses() to wait until any outstanding
     * requests have received a response or error.
     *
     * Use this if you no longer care about getting success responses, but need to do cleanup based
     * on responses for requests that have already been dispatched.
     */
    void interrupt();

    /**
     * Cancels all outstanding requests and makes waitForResponses() return immediately.
     *
     * Use this if you no longer care about getting success responses, and don't need to process
     * responses for outstanding requests.
     */
    void kill();

private:
    /**
     * Returns true if each remote has received a response or error. (If kill() has been called,
     * the error is the error assigned by the TaskExecutor when a callback is canceled).
     */
    bool _done();

    /**
     * Executes the logic of _done().
     */
    bool _done_inlock();

    /**
     * Replaces _notification with a new notification.
     *
     * If _stopRetrying is false, for each remote that does not have a response or outstanding
     * request, schedules work to send the command to the remote.
     *
     * Invalid to call if there is an existing Notification and it has not yet been signaled.
     */
    void _scheduleRequestsIfNeeded(OperationContext* txn);

    /**
     * Helper to schedule a command to a remote.
     *
     * The 'remoteIndex' gives the position of the remote node from which we are retrieving the
     * batch in '_remotes'.
     *
     * Returns success if the command to retrieve the next batch was scheduled successfully.
     */
    Status _scheduleRequest_inlock(OperationContext* txn, size_t remoteIndex);

    /**
     * The callback for a remote command.
     *
     * 'remoteIndex' is the position of the relevant remote node in '_remotes', and therefore
     * indicates which node the response came from and where the response should be buffered.
     *
     * On a retriable error, unless _stopRetrying is true, signals the notification so that the
     * request can be immediately retried.
     *
     * On a non-retriable error, if allowPartialResults is false, sets _stopRetrying to true.
     */
    void _handleResponse(const executor::TaskExecutor::RemoteCommandCallbackArgs& cbData,
                         OperationContext* txn,
                         size_t remoteIndex);

    /**
     * If the existing notification has not yet been signaled, signals it and marks it as signaled.
     */
    void _signalCurrentNotification_inlock();

    /**
     * Wrapper around signalCurrentNotification_inlock(); only signals the notification if _done()
     * is true.
     */
    void _signalCurrentNotificationIfDone_inlock();

    /**
     * We instantiate one of these per remote host.
     */
    struct RemoteData {
        /**
         * Creates a new uninitialized remote state with a command to send.
         */
        RemoteData(ShardId shardId, BSONObj cmdObj);

        /**
         * Returns the resolved host and port on which the remote command was or will be run.
         */
        const HostAndPort& getTargetHost() const;

        /**
         * Given a read preference, selects a host on which the command should be run.
         */
        Status resolveShardIdToHostAndPort(const ReadPreferenceSetting& readPref);

        /**
         * Returns the Shard object associated with this remote.
         */
        std::shared_ptr<Shard> getShard();

        // ShardId of the shard to which the command will be sent.
        const ShardId shardId;

        // The command object to send to the remote host.
        BSONObj cmdObj;

        // The response or error from the remote. Is unset until a response or error has been
        // received.
        boost::optional<StatusWith<executor::RemoteCommandResponse>> swResponse;

        // The exact host on which the remote command was run. Is unset until a request has been
        // sent.
        boost::optional<HostAndPort> shardHostAndPort;

        // The number of times we've retried sending the command to this remote.
        int retryCount = 0;

        // The callback handle to an outstanding request for this remote.
        executor::TaskExecutor::CallbackHandle cbHandle;
    };

    /**
     * Used internally to determine if the ARS should attempt to retry any requests. Is set to true
     * when:
     * - interrupt() or kill() is called
     * - allowPartialResults is false and some remote has a non-retriable error (or exhausts its
     *  retries for a retriable error).
     */
    bool _stopRetrying = false;

    // Not owned here.
    executor::TaskExecutor* _executor;

    // The metadata obj to pass along with the command remote. Used to indicate that the command is
    // ok to run on secondaries.
    BSONObj _metadataObj;

    // The database against which the commands are run.
    const StringData _db;

    // The readPreference to use for all requests.
    ReadPreferenceSetting _readPreference;

    // If set to true, allows for skipping over hosts that have non-retriable errors or exhaust
    // their retries.
    bool _allowPartialResults = false;

    // Must be acquired before accessing any data members.
    // Must also be held when calling any of the '_inlock()' helper functions.
    stdx::mutex _mutex;

    // Data tracking the state of our communication with each of the remote nodes.
    std::vector<RemoteData> _remotes;

    // A notification that gets signaled when a remote has a retriable error or the last outstanding
    // response is received.
    boost::optional<Notification<void>> _notification;
};

}  // namespace mongo