test/core/util/port_server_client.cc - platform/external/grpc-grpc - Git at Google

 /*
  *
  * Copyright 2015 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.
  *
  */

 #include <grpc/support/port_platform.h>
 #include "test/core/util/test_config.h"

 #ifdef GRPC_TEST_PICK_PORT
 #include "test/core/util/port_server_client.h"

 #include <math.h>
 #include <string.h>

 #include <grpc/grpc.h>
 #include <grpc/support/alloc.h>
 #include <grpc/support/log.h>
 #include <grpc/support/string_util.h>
 #include <grpc/support/sync.h>
 #include <grpc/support/time.h>

 #include "src/core/lib/http/httpcli.h"

 typedef struct freereq {
   gpr_mu* mu;
   grpc_polling_entity pops;
   int done;
 } freereq;

 static void destroy_pops_and_shutdown(void* p, grpc_error* error) {
   grpc_pollset* pollset =
       grpc_polling_entity_pollset(static_cast<grpc_polling_entity*>(p));
   grpc_pollset_destroy(pollset);
   gpr_free(pollset);
 }

 static void freed_port_from_server(void* arg, grpc_error* error) {
   freereq* pr = static_cast<freereq*>(arg);
   gpr_mu_lock(pr->mu);
   pr->done = 1;
   GRPC_LOG_IF_ERROR(
       "pollset_kick",
       grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
   gpr_mu_unlock(pr->mu);
 }

 void grpc_free_port_using_server(int port) {
   grpc_httpcli_context context;
   grpc_httpcli_request req;
   grpc_httpcli_response rsp;
   freereq pr;
   char* path;
   grpc_core::ExecCtx exec_ctx;
   grpc_closure* shutdown_closure;

   grpc_init();

   memset(&pr, 0, sizeof(pr));
   memset(&req, 0, sizeof(req));
   memset(&rsp, 0, sizeof(rsp));

   grpc_pollset* pollset =
       static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
   grpc_pollset_init(pollset, &pr.mu);
   pr.pops = grpc_polling_entity_create_from_pollset(pollset);
   shutdown_closure = GRPC_CLOSURE_CREATE(destroy_pops_and_shutdown, &pr.pops,
                                          grpc_schedule_on_exec_ctx);

   req.host = const_cast<char*>(GRPC_PORT_SERVER_ADDRESS);
   gpr_asprintf(&path, "/drop/%d", port);
   req.http.path = path;

   grpc_httpcli_context_init(&context);
   grpc_resource_quota* resource_quota =
       grpc_resource_quota_create("port_server_client/free");
   grpc_httpcli_get(&context, &pr.pops, resource_quota, &req,
                    grpc_core::ExecCtx::Get()->Now() + 30 * GPR_MS_PER_SEC,
                    GRPC_CLOSURE_CREATE(freed_port_from_server, &pr,
                                        grpc_schedule_on_exec_ctx),
                    &rsp);
   grpc_resource_quota_unref_internal(resource_quota);
   grpc_core::ExecCtx::Get()->Flush();
   gpr_mu_lock(pr.mu);
   while (!pr.done) {
     grpc_pollset_worker* worker = nullptr;
     if (!GRPC_LOG_IF_ERROR(
             "pollset_work",
             grpc_pollset_work(
                 grpc_polling_entity_pollset(&pr.pops), &worker,
                 grpc_core::ExecCtx::Get()->Now() + GPR_MS_PER_SEC))) {
       pr.done = 1;
     }
   }
   gpr_mu_unlock(pr.mu);

   grpc_httpcli_context_destroy(&context);
   grpc_pollset_shutdown(grpc_polling_entity_pollset(&pr.pops),
                         shutdown_closure);

   gpr_free(path);
   grpc_http_response_destroy(&rsp);

   grpc_shutdown();
 }

 typedef struct portreq {
   gpr_mu* mu;
   grpc_polling_entity pops;
   int port;
   int retries;
   char* server;
   grpc_httpcli_context* ctx;
   grpc_httpcli_response response;
 } portreq;

 static void got_port_from_server(void* arg, grpc_error* error) {
   size_t i;
   int port = 0;
   portreq* pr = static_cast<portreq*>(arg);
   int failed = 0;
   grpc_httpcli_response* response = &pr->response;

   if (error != GRPC_ERROR_NONE) {
     failed = 1;
     const char* msg = grpc_error_string(error);
     gpr_log(GPR_DEBUG, "failed port pick from server: retrying [%s]", msg);

   } else if (response->status != 200) {
     failed = 1;
     gpr_log(GPR_DEBUG, "failed port pick from server: status=%d",
             response->status);
   }

   if (failed) {
     grpc_httpcli_request req;
     memset(&req, 0, sizeof(req));
     if (pr->retries >= 5) {
       gpr_mu_lock(pr->mu);
       pr->port = 0;
       GRPC_LOG_IF_ERROR(
           "pollset_kick",
           grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
       gpr_mu_unlock(pr->mu);
       return;
     }
     GPR_ASSERT(pr->retries < 10);
     gpr_sleep_until(gpr_time_add(
         gpr_now(GPR_CLOCK_REALTIME),
         gpr_time_from_millis(
             static_cast<int64_t>(
                 1000.0 * (1 + pow(1.3, pr->retries) * rand() / RAND_MAX)),
             GPR_TIMESPAN)));
     pr->retries++;
     req.host = pr->server;
     req.http.path = const_cast<char*>("/get");
     grpc_http_response_destroy(&pr->response);
     memset(&pr->response, 0, sizeof(pr->response));
     grpc_resource_quota* resource_quota =
         grpc_resource_quota_create("port_server_client/pick_retry");
     grpc_httpcli_get(pr->ctx, &pr->pops, resource_quota, &req,
                      grpc_core::ExecCtx::Get()->Now() + 30 * GPR_MS_PER_SEC,
                      GRPC_CLOSURE_CREATE(got_port_from_server, pr,
                                          grpc_schedule_on_exec_ctx),
                      &pr->response);
     grpc_resource_quota_unref_internal(resource_quota);
     return;
   }
   GPR_ASSERT(response);
   GPR_ASSERT(response->status == 200);
   for (i = 0; i < response->body_length; i++) {
     GPR_ASSERT(response->body[i] >= '0' && response->body[i] <= '9');
     port = port * 10 + response->body[i] - '0';
   }
   GPR_ASSERT(port > 1024);
   gpr_mu_lock(pr->mu);
   pr->port = port;
   GRPC_LOG_IF_ERROR(
       "pollset_kick",
       grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
   gpr_mu_unlock(pr->mu);
 }

 int grpc_pick_port_using_server(void) {
   grpc_httpcli_context context;
   grpc_httpcli_request req;
   portreq pr;
   grpc_closure* shutdown_closure;

   grpc_init();
   {
     grpc_core::ExecCtx exec_ctx;
     memset(&pr, 0, sizeof(pr));
     memset(&req, 0, sizeof(req));
     grpc_pollset* pollset =
         static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
     grpc_pollset_init(pollset, &pr.mu);
     pr.pops = grpc_polling_entity_create_from_pollset(pollset);
     shutdown_closure = GRPC_CLOSURE_CREATE(destroy_pops_and_shutdown, &pr.pops,
                                            grpc_schedule_on_exec_ctx);
     pr.port = -1;
     pr.server = const_cast<char*>(GRPC_PORT_SERVER_ADDRESS);
     pr.ctx = &context;

     req.host = const_cast<char*>(GRPC_PORT_SERVER_ADDRESS);
     req.http.path = const_cast<char*>("/get");

     grpc_httpcli_context_init(&context);
     grpc_resource_quota* resource_quota =
         grpc_resource_quota_create("port_server_client/pick");
     grpc_httpcli_get(&context, &pr.pops, resource_quota, &req,
                      grpc_core::ExecCtx::Get()->Now() + 30 * GPR_MS_PER_SEC,
                      GRPC_CLOSURE_CREATE(got_port_from_server, &pr,
                                          grpc_schedule_on_exec_ctx),
                      &pr.response);
     grpc_resource_quota_unref_internal(resource_quota);
     grpc_core::ExecCtx::Get()->Flush();
     gpr_mu_lock(pr.mu);
     while (pr.port == -1) {
       grpc_pollset_worker* worker = nullptr;
       if (!GRPC_LOG_IF_ERROR(
               "pollset_work",
               grpc_pollset_work(
                   grpc_polling_entity_pollset(&pr.pops), &worker,
                   grpc_core::ExecCtx::Get()->Now() + GPR_MS_PER_SEC))) {
         pr.port = 0;
       }
     }
     gpr_mu_unlock(pr.mu);

     grpc_http_response_destroy(&pr.response);
     grpc_httpcli_context_destroy(&context);
     grpc_pollset_shutdown(grpc_polling_entity_pollset(&pr.pops),
                           shutdown_closure);

     grpc_core::ExecCtx::Get()->Flush();
   }
   grpc_shutdown();

   return pr.port;
 }

 #endif  // GRPC_TEST_PICK_PORT
	/*
	*
	* Copyright 2015 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.
	*
	*/

	#include <grpc/support/port_platform.h>
	#include "test/core/util/test_config.h"

	#ifdef GRPC_TEST_PICK_PORT
	#include "test/core/util/port_server_client.h"

	#include <math.h>
	#include <string.h>

	#include <grpc/grpc.h>
	#include <grpc/support/alloc.h>
	#include <grpc/support/log.h>
	#include <grpc/support/string_util.h>
	#include <grpc/support/sync.h>
	#include <grpc/support/time.h>

	#include "src/core/lib/http/httpcli.h"

	typedef struct freereq {
	gpr_mu* mu;
	grpc_polling_entity pops;
	int done;
	} freereq;

	static void destroy_pops_and_shutdown(void* p, grpc_error* error) {
	grpc_pollset* pollset =
	grpc_polling_entity_pollset(static_cast<grpc_polling_entity*>(p));
	grpc_pollset_destroy(pollset);
	gpr_free(pollset);
	}

	static void freed_port_from_server(void* arg, grpc_error* error) {
	freereq* pr = static_cast<freereq*>(arg);
	gpr_mu_lock(pr->mu);
	pr->done = 1;
	GRPC_LOG_IF_ERROR(
	"pollset_kick",
	grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
	gpr_mu_unlock(pr->mu);
	}

	void grpc_free_port_using_server(int port) {
	grpc_httpcli_context context;
	grpc_httpcli_request req;
	grpc_httpcli_response rsp;
	freereq pr;
	char* path;
	grpc_core::ExecCtx exec_ctx;
	grpc_closure* shutdown_closure;

	grpc_init();

	memset(&pr, 0, sizeof(pr));
	memset(&req, 0, sizeof(req));
	memset(&rsp, 0, sizeof(rsp));

	grpc_pollset* pollset =
	static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
	grpc_pollset_init(pollset, &pr.mu);
	pr.pops = grpc_polling_entity_create_from_pollset(pollset);
	shutdown_closure = GRPC_CLOSURE_CREATE(destroy_pops_and_shutdown, &pr.pops,
	grpc_schedule_on_exec_ctx);

	req.host = const_cast<char*>(GRPC_PORT_SERVER_ADDRESS);
	gpr_asprintf(&path, "/drop/%d", port);
	req.http.path = path;

	grpc_httpcli_context_init(&context);
	grpc_resource_quota* resource_quota =
	grpc_resource_quota_create("port_server_client/free");
	grpc_httpcli_get(&context, &pr.pops, resource_quota, &req,
	grpc_core::ExecCtx::Get()->Now() + 30 * GPR_MS_PER_SEC,
	GRPC_CLOSURE_CREATE(freed_port_from_server, &pr,
	grpc_schedule_on_exec_ctx),
	&rsp);
	grpc_resource_quota_unref_internal(resource_quota);
	grpc_core::ExecCtx::Get()->Flush();
	gpr_mu_lock(pr.mu);
	while (!pr.done) {
	grpc_pollset_worker* worker = nullptr;
	if (!GRPC_LOG_IF_ERROR(
	"pollset_work",
	grpc_pollset_work(
	grpc_polling_entity_pollset(&pr.pops), &worker,
	grpc_core::ExecCtx::Get()->Now() + GPR_MS_PER_SEC))) {
	pr.done = 1;
	}
	}
	gpr_mu_unlock(pr.mu);

	grpc_httpcli_context_destroy(&context);
	grpc_pollset_shutdown(grpc_polling_entity_pollset(&pr.pops),
	shutdown_closure);

	gpr_free(path);
	grpc_http_response_destroy(&rsp);

	grpc_shutdown();
	}

	typedef struct portreq {
	gpr_mu* mu;
	grpc_polling_entity pops;
	int port;
	int retries;
	char* server;
	grpc_httpcli_context* ctx;
	grpc_httpcli_response response;
	} portreq;

	static void got_port_from_server(void* arg, grpc_error* error) {
	size_t i;
	int port = 0;
	portreq* pr = static_cast<portreq*>(arg);
	int failed = 0;
	grpc_httpcli_response* response = &pr->response;

	if (error != GRPC_ERROR_NONE) {
	failed = 1;
	const char* msg = grpc_error_string(error);
	gpr_log(GPR_DEBUG, "failed port pick from server: retrying [%s]", msg);

	} else if (response->status != 200) {
	failed = 1;
	gpr_log(GPR_DEBUG, "failed port pick from server: status=%d",
	response->status);
	}

	if (failed) {
	grpc_httpcli_request req;
	memset(&req, 0, sizeof(req));
	if (pr->retries >= 5) {
	gpr_mu_lock(pr->mu);
	pr->port = 0;
	GRPC_LOG_IF_ERROR(
	"pollset_kick",
	grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
	gpr_mu_unlock(pr->mu);
	return;
	}
	GPR_ASSERT(pr->retries < 10);
	gpr_sleep_until(gpr_time_add(
	gpr_now(GPR_CLOCK_REALTIME),
	gpr_time_from_millis(
	static_cast<int64_t>(
	1000.0 * (1 + pow(1.3, pr->retries) * rand() / RAND_MAX)),
	GPR_TIMESPAN)));
	pr->retries++;
	req.host = pr->server;
	req.http.path = const_cast<char*>("/get");
	grpc_http_response_destroy(&pr->response);
	memset(&pr->response, 0, sizeof(pr->response));
	grpc_resource_quota* resource_quota =
	grpc_resource_quota_create("port_server_client/pick_retry");
	grpc_httpcli_get(pr->ctx, &pr->pops, resource_quota, &req,
	grpc_core::ExecCtx::Get()->Now() + 30 * GPR_MS_PER_SEC,
	GRPC_CLOSURE_CREATE(got_port_from_server, pr,
	grpc_schedule_on_exec_ctx),
	&pr->response);
	grpc_resource_quota_unref_internal(resource_quota);
	return;
	}
	GPR_ASSERT(response);
	GPR_ASSERT(response->status == 200);
	for (i = 0; i < response->body_length; i++) {
	GPR_ASSERT(response->body[i] >= '0' && response->body[i] <= '9');
	port = port * 10 + response->body[i] - '0';
	}
	GPR_ASSERT(port > 1024);
	gpr_mu_lock(pr->mu);
	pr->port = port;
	GRPC_LOG_IF_ERROR(
	"pollset_kick",
	grpc_pollset_kick(grpc_polling_entity_pollset(&pr->pops), nullptr));
	gpr_mu_unlock(pr->mu);
	}

	int grpc_pick_port_using_server(void) {
	grpc_httpcli_context context;
	grpc_httpcli_request req;
	portreq pr;
	grpc_closure* shutdown_closure;

	grpc_init();
	{
	grpc_core::ExecCtx exec_ctx;
	memset(&pr, 0, sizeof(pr));
	memset(&req, 0, sizeof(req));
	grpc_pollset* pollset =
	static_cast<grpc_pollset*>(gpr_zalloc(grpc_pollset_size()));
	grpc_pollset_init(pollset, &pr.mu);
	pr.pops = grpc_polling_entity_create_from_pollset(pollset);
	shutdown_closure = GRPC_CLOSURE_CREATE(destroy_pops_and_shutdown, &pr.pops,
	grpc_schedule_on_exec_ctx);
	pr.port = -1;
	pr.server = const_cast<char*>(GRPC_PORT_SERVER_ADDRESS);
	pr.ctx = &context;

	req.host = const_cast<char*>(GRPC_PORT_SERVER_ADDRESS);
	req.http.path = const_cast<char*>("/get");

	grpc_httpcli_context_init(&context);
	grpc_resource_quota* resource_quota =
	grpc_resource_quota_create("port_server_client/pick");
	grpc_httpcli_get(&context, &pr.pops, resource_quota, &req,
	grpc_core::ExecCtx::Get()->Now() + 30 * GPR_MS_PER_SEC,
	GRPC_CLOSURE_CREATE(got_port_from_server, &pr,
	grpc_schedule_on_exec_ctx),
	&pr.response);
	grpc_resource_quota_unref_internal(resource_quota);
	grpc_core::ExecCtx::Get()->Flush();
	gpr_mu_lock(pr.mu);
	while (pr.port == -1) {
	grpc_pollset_worker* worker = nullptr;
	if (!GRPC_LOG_IF_ERROR(
	"pollset_work",
	grpc_pollset_work(
	grpc_polling_entity_pollset(&pr.pops), &worker,
	grpc_core::ExecCtx::Get()->Now() + GPR_MS_PER_SEC))) {
	pr.port = 0;
	}
	}
	gpr_mu_unlock(pr.mu);

	grpc_http_response_destroy(&pr.response);
	grpc_httpcli_context_destroy(&context);
	grpc_pollset_shutdown(grpc_polling_entity_pollset(&pr.pops),
	shutdown_closure);

	grpc_core::ExecCtx::Get()->Flush();
	}
	grpc_shutdown();

	return pr.port;
	}

	#endif // GRPC_TEST_PICK_PORT