|
|
|
// Copyright 2016 gRPC authors.
|
|
|
|
//
|
|
|
|
// Licensed 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 GRPC_TEST_CPP_END2END_CONNECTION_DELAY_INJECTOR_H
|
|
|
|
#define GRPC_TEST_CPP_END2END_CONNECTION_DELAY_INJECTOR_H
|
|
|
|
|
|
|
|
#include <memory>
|
|
|
|
|
|
|
|
#include "src/core/lib/event_engine/channel_args_endpoint_config.h"
|
|
|
|
#include "src/core/lib/gprpp/time.h"
|
|
|
|
#include "src/core/lib/iomgr/socket_utils_posix.h"
|
|
|
|
#include "src/core/lib/iomgr/tcp_client.h"
|
|
|
|
#include "src/core/lib/iomgr/timer.h"
|
|
|
|
|
|
|
|
namespace grpc {
|
|
|
|
namespace testing {
|
|
|
|
|
|
|
|
// Allows injecting connection-establishment delays into C-core.
|
|
|
|
// Typical usage:
|
|
|
|
//
|
|
|
|
// // At grpc_init() time.
|
|
|
|
// ConnectionAttemptInjector::Init();
|
|
|
|
//
|
|
|
|
// // When an injection is desired.
|
|
|
|
// ConnectionDelayInjector delay_injector(grpc_core::Duration::Seconds(10));
|
|
|
|
// delay_injector.Start();
|
|
|
|
//
|
|
|
|
// The injection is global, so there must be only one ConnectionAttemptInjector
|
|
|
|
// object at any one time.
|
|
|
|
class ConnectionAttemptInjector {
|
|
|
|
public:
|
|
|
|
// Global initializer. Replaces the iomgr TCP client vtable.
|
|
|
|
// Must be called exactly once after grpc_init() but before any TCP
|
|
|
|
// connections are established.
|
|
|
|
static void Init();
|
|
|
|
|
|
|
|
virtual ~ConnectionAttemptInjector();
|
|
|
|
|
|
|
|
// Must be called after instantiation.
|
|
|
|
void Start();
|
|
|
|
|
|
|
|
// Invoked for every TCP connection attempt.
|
|
|
|
// Implementations must eventually either invoke the closure
|
|
|
|
// themselves or delegate to the iomgr implementation by calling
|
|
|
|
// AttemptConnection(). QueuedAttempt may be used to queue an attempt
|
|
|
|
// for asynchronous processing.
|
|
|
|
virtual void HandleConnection(
|
|
|
|
grpc_closure* closure, grpc_endpoint** ep,
|
|
|
|
grpc_pollset_set* interested_parties,
|
|
|
|
const grpc_event_engine::experimental::EndpointConfig& config,
|
|
|
|
const grpc_resolved_address* addr, grpc_core::Timestamp deadline) = 0;
|
|
|
|
|
|
|
|
protected:
|
|
|
|
// Represents a queued attempt.
|
|
|
|
// The caller must invoke either Resume() or Fail() before destroying.
|
|
|
|
class QueuedAttempt {
|
|
|
|
public:
|
|
|
|
QueuedAttempt(grpc_closure* closure, grpc_endpoint** ep,
|
|
|
|
grpc_pollset_set* interested_parties,
|
|
|
|
const grpc_event_engine::experimental::EndpointConfig& config,
|
|
|
|
const grpc_resolved_address* addr,
|
|
|
|
grpc_core::Timestamp deadline)
|
|
|
|
: closure_(closure),
|
|
|
|
endpoint_(ep),
|
|
|
|
interested_parties_(interested_parties),
|
|
|
|
config_(*reinterpret_cast<const grpc_event_engine::experimental::
|
|
|
|
ChannelArgsEndpointConfig*>(&config)),
|
|
|
|
deadline_(deadline) {
|
|
|
|
memcpy(&address_, addr, sizeof(address_));
|
|
|
|
}
|
|
|
|
|
|
|
|
~QueuedAttempt() { GPR_ASSERT(closure_ == nullptr); }
|
|
|
|
|
|
|
|
// Caller must invoke this from a thread with an ExecCtx.
|
|
|
|
void Resume() {
|
|
|
|
GPR_ASSERT(closure_ != nullptr);
|
|
|
|
AttemptConnection(closure_, endpoint_, interested_parties_, config_,
|
|
|
|
&address_, deadline_);
|
|
|
|
closure_ = nullptr;
|
|
|
|
}
|
|
|
|
|
|
|
|
// Caller must invoke this from a thread with an ExecCtx.
|
|
|
|
void Fail(grpc_error_handle error) {
|
|
|
|
GPR_ASSERT(closure_ != nullptr);
|
|
|
|
grpc_core::ExecCtx::Run(DEBUG_LOCATION, closure_, error);
|
|
|
|
closure_ = nullptr;
|
|
|
|
}
|
|
|
|
|
|
|
|
private:
|
|
|
|
grpc_closure* closure_;
|
|
|
|
grpc_endpoint** endpoint_;
|
|
|
|
grpc_pollset_set* interested_parties_;
|
|
|
|
grpc_event_engine::experimental::ChannelArgsEndpointConfig config_;
|
|
|
|
grpc_resolved_address address_;
|
|
|
|
grpc_core::Timestamp deadline_;
|
|
|
|
};
|
|
|
|
|
|
|
|
// Injects a delay before continuing a connection attempt.
|
|
|
|
class InjectedDelay {
|
|
|
|
public:
|
|
|
|
virtual ~InjectedDelay() = default;
|
|
|
|
|
|
|
|
InjectedDelay(grpc_core::Duration duration, grpc_closure* closure,
|
|
|
|
grpc_endpoint** ep, grpc_pollset_set* interested_parties,
|
|
|
|
const grpc_event_engine::experimental::EndpointConfig& config,
|
|
|
|
const grpc_resolved_address* addr,
|
|
|
|
grpc_core::Timestamp deadline);
|
|
|
|
|
|
|
|
private:
|
|
|
|
// Subclasses can override to perform an action when the attempt resumes.
|
|
|
|
virtual void BeforeResumingAction() {}
|
|
|
|
|
|
|
|
static void TimerCallback(void* arg, grpc_error_handle /*error*/);
|
|
|
|
|
|
|
|
QueuedAttempt attempt_;
|
|
|
|
grpc_timer timer_;
|
|
|
|
grpc_closure timer_callback_;
|
|
|
|
};
|
|
|
|
|
|
|
|
static void AttemptConnection(
|
|
|
|
grpc_closure* closure, grpc_endpoint** ep,
|
|
|
|
grpc_pollset_set* interested_parties,
|
|
|
|
const grpc_event_engine::experimental::EndpointConfig& config,
|
|
|
|
const grpc_resolved_address* addr, grpc_core::Timestamp deadline);
|
|
|
|
};
|
|
|
|
|
|
|
|
// A concrete implementation that injects a fixed delay.
|
|
|
|
class ConnectionDelayInjector : public ConnectionAttemptInjector {
|
|
|
|
public:
|
|
|
|
explicit ConnectionDelayInjector(grpc_core::Duration duration)
|
|
|
|
: duration_(duration) {}
|
|
|
|
|
|
|
|
void HandleConnection(
|
|
|
|
grpc_closure* closure, grpc_endpoint** ep,
|
|
|
|
grpc_pollset_set* interested_parties,
|
|
|
|
const grpc_event_engine::experimental::EndpointConfig& config,
|
|
|
|
const grpc_resolved_address* addr,
|
|
|
|
grpc_core::Timestamp deadline) override;
|
|
|
|
|
|
|
|
private:
|
|
|
|
grpc_core::Duration duration_;
|
|
|
|
};
|
|
|
|
|
|
|
|
// A concrete implementation that allows injecting holds for individual
|
|
|
|
// connection attemps, one at a time.
|
|
|
|
class ConnectionHoldInjector : public ConnectionAttemptInjector {
|
|
|
|
private:
|
|
|
|
grpc_core::Mutex mu_; // Needs to be declared up front.
|
|
|
|
|
|
|
|
public:
|
|
|
|
class Hold {
|
|
|
|
public:
|
|
|
|
// Do not instantiate directly -- must be created via AddHold().
|
|
|
|
Hold(ConnectionHoldInjector* injector, int port, bool intercept_completion);
|
|
|
|
|
|
|
|
// Waits for the connection attempt to start.
|
|
|
|
// After this returns, exactly one of Resume() or Fail() must be called.
|
|
|
|
void Wait();
|
|
|
|
|
|
|
|
// Resumes a connection attempt. Must be called after Wait().
|
|
|
|
void Resume();
|
|
|
|
|
|
|
|
// Fails a connection attempt. Must be called after Wait().
|
|
|
|
void Fail(grpc_error_handle error);
|
|
|
|
|
|
|
|
// If the hold was created with intercept_completion=true, then this
|
|
|
|
// can be called after Resume() to wait for the connection attempt
|
|
|
|
// to complete.
|
|
|
|
void WaitForCompletion();
|
|
|
|
|
|
|
|
// Returns true if the connection attempt has been started.
|
|
|
|
bool IsStarted();
|
|
|
|
|
|
|
|
private:
|
|
|
|
friend class ConnectionHoldInjector;
|
|
|
|
|
|
|
|
static void OnComplete(void* arg, grpc_error_handle error);
|
|
|
|
|
|
|
|
ConnectionHoldInjector* injector_;
|
|
|
|
const int port_;
|
|
|
|
const bool intercept_completion_;
|
|
|
|
std::unique_ptr<QueuedAttempt> queued_attempt_
|
|
|
|
ABSL_GUARDED_BY(&ConnectionHoldInjector::mu_);
|
|
|
|
grpc_core::CondVar start_cv_;
|
|
|
|
grpc_closure on_complete_;
|
|
|
|
grpc_closure* original_on_complete_;
|
|
|
|
grpc_core::CondVar complete_cv_;
|
|
|
|
};
|
|
|
|
|
|
|
|
// Adds a hold for a given port. The caller may then use Wait() on
|
|
|
|
// the resulting Hold object to wait for the connection attempt to start.
|
|
|
|
// If intercept_completion is true, the caller can use WaitForCompletion()
|
|
|
|
// on the resulting Hold object.
|
|
|
|
std::unique_ptr<Hold> AddHold(int port, bool intercept_completion = false);
|
|
|
|
|
|
|
|
private:
|
|
|
|
void HandleConnection(
|
|
|
|
grpc_closure* closure, grpc_endpoint** ep,
|
|
|
|
grpc_pollset_set* interested_parties,
|
|
|
|
const grpc_event_engine::experimental::EndpointConfig& config,
|
|
|
|
const grpc_resolved_address* addr,
|
|
|
|
grpc_core::Timestamp deadline) override;
|
|
|
|
|
|
|
|
std::vector<Hold*> holds_;
|
|
|
|
};
|
|
|
|
|
|
|
|
} // namespace testing
|
|
|
|
} // namespace grpc
|
|
|
|
|
|
|
|
#endif // GRPC_TEST_CPP_END2END_CONNECTION_DELAY_INJECTOR_H
|