natas_hw
/
grpc
0
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'hybrid' of github.com:ctiller/grpc into hybrid

This commit is contained in:
Craig Tiller 2017-04-12 14:02:19 -07:00
parent b3612d3cd4 11c5832b3e
commit dea325b6e4
15 changed files with 169 additions and 111 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
include/grpc++/impl/codegen/client_unary_call.h
View File

@ -52,7 +52,9 @@ template <class InputMessage, class OutputMessage>
Status BlockingUnaryCall(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const InputMessage& request,
OutputMessage* result) {
CompletionQueue cq(true); // Pluckable completion queue
CompletionQueue cq(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}); // Pluckable completion queue
Call call(channel->CreateCall(method, context, &cq));
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpRecvInitialMetadata, CallOpRecvMessage<OutputMessage>,

32
include/grpc++/impl/codegen/completion_queue.h
View File

@ -102,7 +102,9 @@ class CompletionQueue : private GrpcLibraryCodegen {
public:
/// Default constructor. Implicitly creates a \a grpc_completion_queue
/// instance.
CompletionQueue() : CompletionQueue(false) {}
CompletionQueue()
: CompletionQueue(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, GRPC_CQ_DEFAULT_POLLING}) {}
/// Wrap \a take, taking ownership of the instance.
///
@ -182,6 +184,16 @@ class CompletionQueue : private GrpcLibraryCodegen {
};
void CompleteAvalanching();
protected:
/// Private constructor of CompletionQueue only visible to friend classes
CompletionQueue(const grpc_completion_queue_attributes& attributes) {
cq_ = g_core_codegen_interface->grpc_completion_queue_create(
g_core_codegen_interface->grpc_completion_queue_factory_lookup(
&attributes),
&attributes, NULL);
InitialAvalanching(); // reserve this for the future shutdown
}
private:
// Friend synchronous wrappers so that they can access Pluck(), which is
// a semi-private API geared towards the synchronous implementation.
@ -215,18 +227,6 @@ class CompletionQueue : private GrpcLibraryCodegen {
const InputMessage& request,
OutputMessage* result);
/// Private constructor of CompletionQueue only visible to friend classes
CompletionQueue(bool is_pluck) {
if (is_pluck) {
cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_pluck(
nullptr);
} else {
cq_ = g_core_codegen_interface->grpc_completion_queue_create_for_next(
nullptr);
}
InitialAvalanching(); // reserve this for the future shutdown
}
NextStatus AsyncNextInternal(void** tag, bool* ok, gpr_timespec deadline);
/// Wraps \a grpc_completion_queue_pluck.
@ -299,11 +299,9 @@ class ServerCompletionQueue : public CompletionQueue {
/// AsyncNext()). By default all server completion queues are assumed to be
/// frequently polled.
ServerCompletionQueue(grpc_cq_polling_type polling_type)
: CompletionQueue(MakeCompletionQueue(polling_type)),
: CompletionQueue(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_NEXT, polling_type}),
polling_type_(polling_type) {}
static grpc_completion_queue* MakeCompletionQueue(
grpc_cq_polling_type polling_type);
};
} // namespace grpc

9
include/grpc++/impl/codegen/core_codegen.h
View File

@ -44,8 +44,15 @@
namespace grpc {
/// Implementation of the core codegen interface.
class CoreCodegen : public CoreCodegenInterface {
class CoreCodegen final : public CoreCodegenInterface {
private:
virtual const grpc_completion_queue_factory*
grpc_completion_queue_factory_lookup(
const grpc_completion_queue_attributes* attributes) override;
virtual grpc_completion_queue* grpc_completion_queue_create(
const grpc_completion_queue_factory* factory,
const grpc_completion_queue_attributes* attributes,
void* reserved) override;
grpc_completion_queue* grpc_completion_queue_create_for_next(
void* reserved) override;
grpc_completion_queue* grpc_completion_queue_create_for_pluck(

6
include/grpc++/impl/codegen/core_codegen_interface.h
View File

@ -59,6 +59,12 @@ class CoreCodegenInterface {
virtual void assert_fail(const char* failed_assertion, const char* file,
int line) = 0;
virtual const grpc_completion_queue_factory*
grpc_completion_queue_factory_lookup(
const grpc_completion_queue_attributes* attributes) = 0;
virtual grpc_completion_queue* grpc_completion_queue_create(
const grpc_completion_queue_factory* factory,
const grpc_completion_queue_attributes* attributes, void* reserved) = 0;
virtual grpc_completion_queue* grpc_completion_queue_create_for_next(
void* reserved) = 0;
virtual grpc_completion_queue* grpc_completion_queue_create_for_pluck(

12
include/grpc++/impl/codegen/sync_stream.h
View File

@ -156,7 +156,9 @@ class ClientReader final : public ClientReaderInterface<R> {
ClientReader(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, const W& request)
: context_(context),
cq_(true), // Pluckable cq
cq_(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq
call_(channel->CreateCall(method, context, &cq_)) {
CallOpSet<CallOpSendInitialMetadata, CallOpSendMessage,
CallOpClientSendClose>
@ -230,7 +232,9 @@ class ClientWriter : public ClientWriterInterface<W> {
ClientWriter(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context, R* response)
: context_(context),
cq_(true), // Pluckable cq
cq_(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq
call_(channel->CreateCall(method, context, &cq_)) {
finish_ops_.RecvMessage(response);
finish_ops_.AllowNoMessage();
@ -330,7 +334,9 @@ class ClientReaderWriter final : public ClientReaderWriterInterface<W, R> {
ClientReaderWriter(ChannelInterface* channel, const RpcMethod& method,
ClientContext* context)
: context_(context),
cq_(true), // Pluckable cq
cq_(grpc_completion_queue_attributes{
GRPC_CQ_CURRENT_VERSION, GRPC_CQ_PLUCK,
GRPC_CQ_DEFAULT_POLLING}), // Pluckable cq
call_(channel->CreateCall(method, context, &cq_)) {
if (!context_->initial_metadata_corked_) {
CallOpSet<CallOpSendInitialMetadata> ops;

23
include/grpc/grpc.h
View File

@ -93,29 +93,6 @@ GRPCAPI const char *grpc_version_string(void);
/** Return a string specifying what the 'g' in gRPC stands for */
GRPCAPI const char *grpc_g_stands_for(void);
/** Specifies the type of APIs to use to pop events from the completion queue */
typedef enum {
/** Events are popped out by calling grpc_completion_queue_next() API ONLY */
GRPC_CQ_NEXT = 1,
/** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/
GRPC_CQ_PLUCK
} grpc_cq_completion_type;
#define GRPC_CQ_CURRENT_VERSION 1
typedef struct grpc_completion_queue_attributes {
/* The version number of this structure. More fields might be added to this
structure in future. */
int version; /* Set to GRPC_CQ_CURRENT_VERSION */
grpc_cq_completion_type cq_completion_type;
grpc_cq_polling_type cq_polling_type;
} grpc_completion_queue_attributes;
/** The completion queue factory structure is opaque to the callers of grpc */
typedef struct grpc_completion_queue_factory grpc_completion_queue_factory;
/** Returns the completion queue factory based on the attributes. MAY return a
NULL if no factory can be found */
GRPCAPI const grpc_completion_queue_factory *

23
include/grpc/impl/codegen/grpc_types.h
View File

@ -569,6 +569,29 @@ typedef enum {
GRPC_CQ_NON_POLLING
} grpc_cq_polling_type;
/** Specifies the type of APIs to use to pop events from the completion queue */
typedef enum {
/** Events are popped out by calling grpc_completion_queue_next() API ONLY */
GRPC_CQ_NEXT = 1,
/** Events are popped out by calling grpc_completion_queue_pluck() API ONLY*/
GRPC_CQ_PLUCK
} grpc_cq_completion_type;
#define GRPC_CQ_CURRENT_VERSION 1
typedef struct grpc_completion_queue_attributes {
/* The version number of this structure. More fields might be added to this
structure in future. */
int version; /* Set to GRPC_CQ_CURRENT_VERSION */
grpc_cq_completion_type cq_completion_type;
grpc_cq_polling_type cq_polling_type;
} grpc_completion_queue_attributes;
/** The completion queue factory structure is opaque to the callers of grpc */
typedef struct grpc_completion_queue_factory grpc_completion_queue_factory;
#ifdef __cplusplus
}
#endif

33
src/core/lib/surface/completion_queue.c
View File

@ -61,6 +61,8 @@ typedef struct {
} plucker;
typedef struct {
bool can_get_pollset;
bool can_listen;
size_t (*size)(void);
void (*init)(grpc_pollset *pollset, gpr_mu **mu);
grpc_error *(*kick)(grpc_pollset *pollset,
@ -107,9 +109,10 @@ static grpc_error *non_polling_poller_work(grpc_exec_ctx *exec_ctx,
gpr_timespec now,
gpr_timespec deadline) {
non_polling_poller *npp = (non_polling_poller *)pollset;
if (npp->shutdown) return GRPC_ERROR_NONE;
non_polling_worker w;
gpr_cv_init(&w.cv);
*worker = (grpc_pollset_worker *)&w;
if (worker != NULL) *worker = (grpc_pollset_worker *)&w;
if (npp->root == NULL) {
npp->root = w.next = w.prev = &w;
} else {
@ -128,11 +131,11 @@ static grpc_error *non_polling_poller_work(grpc_exec_ctx *exec_ctx,
}
npp->root = NULL;
}
w.next->prev = w.prev;
w.prev->next = w.next;
}
w.next->prev = w.prev;
w.prev->next = w.next;
gpr_cv_destroy(&w.cv);
*worker = NULL;
if (worker != NULL) *worker = NULL;
return GRPC_ERROR_NONE;
}
@ -169,21 +172,27 @@ static void non_polling_poller_shutdown(grpc_exec_ctx *exec_ctx,
static const cq_poller_vtable g_poller_vtable_by_poller_type[] = {
/* GRPC_CQ_DEFAULT_POLLING */
{.size = grpc_pollset_size,
{.can_get_pollset = true,
.can_listen = true,
.size = grpc_pollset_size,
.init = grpc_pollset_init,
.kick = grpc_pollset_kick,
.work = grpc_pollset_work,
.shutdown = grpc_pollset_shutdown,
.destroy = grpc_pollset_destroy},
/* GRPC_CQ_NON_LISTENING */
{.size = grpc_pollset_size,
{.can_get_pollset = true,
.can_listen = false,
.size = grpc_pollset_size,
.init = grpc_pollset_init,
.kick = grpc_pollset_kick,
.work = grpc_pollset_work,
.shutdown = grpc_pollset_shutdown,
.destroy = grpc_pollset_destroy},
/* GRPC_CQ_NON_POLLING */
{.size = non_polling_poller_size,
{.can_get_pollset = false,
.can_listen = false,
.size = non_polling_poller_size,
.init = non_polling_poller_init,
.kick = non_polling_poller_kick,
.work = non_polling_poller_work,
@ -837,7 +846,7 @@ void grpc_completion_queue_destroy(grpc_completion_queue *cc) {
}
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc) {
return POLLSET_FROM_CQ(cc);
return cc->poller_vtable->can_get_pollset ? POLLSET_FROM_CQ(cc) : NULL;
}
grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps) {
@ -858,4 +867,10 @@ bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc) {
void grpc_cq_mark_server_cq(grpc_completion_queue *cc) { cc->is_server_cq = 1; }
int grpc_cq_is_server_cq(grpc_completion_queue *cc) { return cc->is_server_cq; }
bool grpc_cq_is_server_cq(grpc_completion_queue *cc) {
return cc->is_server_cq;
}
bool grpc_cq_can_listen(grpc_completion_queue *cc) {
return cc->poller_vtable->can_listen;
}

5
src/core/lib/surface/completion_queue.h
View File

@ -94,10 +94,9 @@ void grpc_cq_end_op(grpc_exec_ctx *exec_ctx, grpc_completion_queue *cc,
grpc_pollset *grpc_cq_pollset(grpc_completion_queue *cc);
grpc_completion_queue *grpc_cq_from_pollset(grpc_pollset *ps);
void grpc_cq_mark_non_listening_server_cq(grpc_completion_queue *cc);
bool grpc_cq_is_non_listening_server_cq(grpc_completion_queue *cc);
void grpc_cq_mark_server_cq(grpc_completion_queue *cc);
int grpc_cq_is_server_cq(grpc_completion_queue *cc);
bool grpc_cq_is_server_cq(grpc_completion_queue *cc);
bool grpc_cq_can_listen(grpc_completion_queue *cc);
grpc_cq_completion_type grpc_get_cq_completion_type(grpc_completion_queue *cc);

19
src/core/lib/surface/server.c
View File

@ -974,7 +974,7 @@ const grpc_channel_filter grpc_server_top_filter = {
static void register_completion_queue(grpc_server *server,
grpc_completion_queue *cq,
bool is_non_listening, void *reserved) {
void *reserved) {
size_t i, n;
GPR_ASSERT(!reserved);
for (i = 0; i < server->cq_count; i++) {
@ -983,10 +983,6 @@ static void register_completion_queue(grpc_server *server,
grpc_cq_mark_server_cq(cq);
if (is_non_listening) {
grpc_cq_mark_non_listening_server_cq(cq);
}
GRPC_CQ_INTERNAL_REF(cq, "server");
n = server->cq_count++;
server->cqs = gpr_realloc(server->cqs,
@ -1009,16 +1005,7 @@ void grpc_server_register_completion_queue(grpc_server *server,
calls grpc_completion_queue_pluck() on server completion queues */
}
register_completion_queue(server, cq, false, reserved);
}
void grpc_server_register_non_listening_completion_queue(
grpc_server *server, grpc_completion_queue *cq, void *reserved) {
GRPC_API_TRACE(
"grpc_server_register_non_listening_completion_queue(server=%p, cq=%p, "
"reserved=%p)",
3, (server, cq, reserved));
register_completion_queue(server, cq, true, reserved);
register_completion_queue(server, cq, reserved);
}
grpc_server *grpc_server_create(const grpc_channel_args *args, void *reserved) {
@ -1101,7 +1088,7 @@ void grpc_server_start(grpc_server *server) {
server->requested_calls_per_cq =
gpr_malloc(sizeof(*server->requested_calls_per_cq) * server->cq_count);
for (i = 0; i < server->cq_count; i++) {
if (!grpc_cq_is_non_listening_server_cq(server->cqs[i])) {
if (grpc_cq_can_listen(server->cqs[i])) {
server->pollsets[server->pollset_count++] =
grpc_cq_pollset(server->cqs[i]);
}

12
src/cpp/common/core_codegen.cc
View File

@ -54,6 +54,18 @@ struct grpc_byte_buffer;
namespace grpc {
const grpc_completion_queue_factory*
CoreCodegen::grpc_completion_queue_factory_lookup(
const grpc_completion_queue_attributes* attributes) {
return ::grpc_completion_queue_factory_lookup(attributes);
}
grpc_completion_queue* CoreCodegen::grpc_completion_queue_create(
const grpc_completion_queue_factory* factory,
const grpc_completion_queue_attributes* attributes, void* reserved) {
return ::grpc_completion_queue_create(factory, attributes, reserved);
}
grpc_completion_queue* CoreCodegen::grpc_completion_queue_create_for_next(
void* reserved) {
return ::grpc_completion_queue_create_for_next(reserved);

26
src/cpp/server/server_builder.cc
View File

@ -243,6 +243,16 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
sync_server_cqs(std::make_shared<
std::vector<std::unique_ptr<ServerCompletionQueue>>>());
int num_frequently_polled_cqs = 0;
for (auto it = cqs_.begin(); it != cqs_.end(); ++it) {
if ((*it)->IsFrequentlyPolled()) {
num_frequently_polled_cqs++;
}
}
const bool is_hybrid_server =
has_sync_methods && num_frequently_polled_cqs > 0;
if (has_sync_methods) {
// This is a Sync server
gpr_log(GPR_INFO,
@ -253,7 +263,7 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
sync_server_settings_.cq_timeout_msec);
grpc_cq_polling_type polling_type =
cqs_.empty() ? GRPC_CQ_DEFAULT_POLLING : GRPC_CQ_NON_POLLING;
is_hybrid_server ? GRPC_CQ_NON_POLLING : GRPC_CQ_DEFAULT_POLLING;
// Create completion queues to listen to incoming rpc requests
for (int i = 0; i < sync_server_settings_.num_cqs; i++) {
@ -273,12 +283,10 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
// server
// 2. cqs_: Completion queues added via AddCompletionQueue() call
// All sync cqs (if any) are frequently polled by ThreadManager
int num_frequently_polled_cqs = sync_server_cqs->size();
for (auto it = sync_server_cqs->begin(); it != sync_server_cqs->end(); ++it) {
grpc_server_register_completion_queue(server->server_, (*it)->cq(),
nullptr);
num_frequently_polled_cqs++;
}
// cqs_ contains the completion queue added by calling the ServerBuilder's
@ -287,14 +295,8 @@ std::unique_ptr<Server> ServerBuilder::BuildAndStart() {
// listening to incoming channels. Such completion queues must be registered
// as non-listening queues
for (auto it = cqs_.begin(); it != cqs_.end(); ++it) {
if ((*it)->IsFrequentlyPolled()) {
grpc_server_register_completion_queue(server->server_, (*it)->cq(),
nullptr);
num_frequently_polled_cqs++;
} else {
grpc_server_register_non_listening_completion_queue(server->server_,
(*it)->cq(), nullptr);
}
grpc_server_register_completion_queue(server->server_, (*it)->cq(),
nullptr);
}
if (num_frequently_polled_cqs == 0) {

46
src/ruby/spec/generic/rpc_server_pool_spec.rb
View File

@ -52,28 +52,31 @@ describe GRPC::Pool do
expect(p.ready_for_work?).to be(false)
end
it 'it stops being ready after all workers jobs waiting or running' do
it 'it stops being ready after all workers are busy' do
p = Pool.new(5)
p.start
job = proc { sleep(3) } # sleep so workers busy when done scheduling
5.times do
expect(p.ready_for_work?).to be(true)
p.schedule(&job)
end
expect(p.ready_for_work?).to be(false)
end
it 'it becomes ready again after jobs complete' do
p = Pool.new(5)
p.start
job = proc {}
wait_mu = Mutex.new
wait_cv = ConditionVariable.new
wait = true
job = proc do
wait_mu.synchronize do
wait_cv.wait(wait_mu) while wait
end
end
5.times do
expect(p.ready_for_work?).to be(true)
p.schedule(&job)
end
expect(p.ready_for_work?).to be(false)
sleep 5 # give the pool time do get at least one task done
expect(p.ready_for_work?).to be(true)
wait_mu.synchronize do
wait = false
wait_cv.broadcast
end
end
end
@ -105,13 +108,20 @@ describe GRPC::Pool do
it 'stops jobs when there are long running jobs' do
p = Pool.new(1)
p.start
o, q = Object.new, Queue.new
wait_forever_mu = Mutex.new
wait_forever_cv = ConditionVariable.new
wait_forever = true
job_running = Queue.new
job = proc do
sleep(5) # long running
q.push(o)
job_running.push(Object.new)
wait_forever_mu.synchronize do
wait_forever_cv.wait while wait_forever
end
end
p.schedule(&job)
sleep(1) # should ensure the long job gets scheduled
job_running.pop
expect { p.stop }.not_to raise_error
end
end

2
test/core/security/credentials_test.c
View File

@ -582,7 +582,7 @@ static void on_oauth2_creds_get_metadata_failure(
static void validate_compute_engine_http_request(
const grpc_httpcli_request *request) {
GPR_ASSERT(request->handshaker != &grpc_httpcli_ssl);
GPR_ASSERT(strcmp(request->host, "metadata") == 0);
GPR_ASSERT(strcmp(request->host, "metadata.google.internal") == 0);
GPR_ASSERT(
strcmp(request->http.path,
"/computeMetadata/v1/instance/service-accounts/default/token") ==

28
test/cpp/end2end/async_end2end_test.cc
View File

@ -38,6 +38,7 @@
#include <grpc++/channel.h>
#include <grpc++/client_context.h>
#include <grpc++/create_channel.h>
#include <grpc++/ext/health_check_service_server_builder_option.h>
#include <grpc++/server.h>
#include <grpc++/server_builder.h>
#include <grpc++/server_context.h>
@ -49,6 +50,7 @@
#include <gtest/gtest.h>
#include "src/core/lib/iomgr/port.h"
#include "src/proto/grpc/health/v1/health.grpc.pb.h"
#include "src/proto/grpc/testing/duplicate/echo_duplicate.grpc.pb.h"
#include "src/proto/grpc/testing/echo.grpc.pb.h"
#include "test/core/util/port.h"
@ -224,13 +226,15 @@ class ServerBuilderSyncPluginDisabler : public ::grpc::ServerBuilderOption {
class TestScenario {
public:
TestScenario(bool non_block, const grpc::string& creds_type,
TestScenario(bool non_block, const grpc::string& creds_type, bool hcs,
const grpc::string& content)
: disable_blocking(non_block),
health_check_service(hcs),
credentials_type(creds_type),
message_content(content) {}
void Log() const;
bool disable_blocking;
bool health_check_service;
// Although the below grpc::string's are logically const, we can't declare
// them const because of a limitation in the way old compilers (e.g., gcc-4.4)
// manage vector insertion using a copy constructor
@ -243,6 +247,8 @@ static std::ostream& operator<<(std::ostream& out,
return out << "TestScenario{disable_blocking="
<< (scenario.disable_blocking ? "true" : "false")
<< ", credentials='" << scenario.credentials_type
<< ", health_check_service="
<< (scenario.health_check_service ? "true" : "false")
<< "', message_size=" << scenario.message_content.size() << "}";
}
@ -252,6 +258,8 @@ void TestScenario::Log() const {
gpr_log(GPR_DEBUG, "%s", out.str().c_str());
}
class HealthCheck : public health::v1::Health::Service {};
class AsyncEnd2endTest : public ::testing::TestWithParam<TestScenario> {
protected:
AsyncEnd2endTest() { GetParam().Log(); }
@ -268,6 +276,9 @@ class AsyncEnd2endTest : public ::testing::TestWithParam<TestScenario> {
GetParam().credentials_type);
builder.AddListeningPort(server_address_.str(), server_creds);
builder.RegisterService(&service_);
if (GetParam().health_check_service) {
builder.RegisterService(&health_check_);
}
cq_ = builder.AddCompletionQueue();
// TODO(zyc): make a test option to choose wheather sync plugins should be
@ -340,6 +351,7 @@ class AsyncEnd2endTest : public ::testing::TestWithParam<TestScenario> {
std::unique_ptr<grpc::testing::EchoTestService::Stub> stub_;
std::unique_ptr<Server> server_;
grpc::testing::EchoTestService::AsyncService service_;
HealthCheck health_check_;
std::ostringstream server_address_;
int port_;
@ -1754,12 +1766,14 @@ std::vector<TestScenario> CreateTestScenarios(bool test_disable_blocking,
messages.push_back(big_msg);
}
for (auto cred = credentials_types.begin(); cred != credentials_types.end();
++cred) {
for (auto msg = messages.begin(); msg != messages.end(); msg++) {
scenarios.emplace_back(false, *cred, *msg);
if (test_disable_blocking) {
scenarios.emplace_back(true, *cred, *msg);
for (auto health_check_service : {false, true}) {
for (auto cred = credentials_types.begin(); cred != credentials_types.end();
++cred) {
for (auto msg = messages.begin(); msg != messages.end(); msg++) {
scenarios.emplace_back(false, *cred, health_check_service, *msg);
if (test_disable_blocking) {
scenarios.emplace_back(true, *cred, health_check_service, *msg);
}
}
}
}