Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Queue batch prediction tasks collectively #753

Open
wants to merge 3 commits into
base: develop
Choose a base branch
from
Open

Queue batch prediction tasks collectively #753

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
5b19a56
Queue batch prediction tasks collectively
robberlang Oct 30, 2019
ed48164
Handle case when no active containers found properly
robberlang Dec 1, 2019
2774441
Handle case when no active containers found properly
robberlang Aug 29, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 41 additions & 1 deletion integration-tests/clipper_admin_tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -791,6 +791,45 @@ def predict_func(inputs):
self.assertGreaterEqual(num_max_batch_queries,
int(total_num_queries * .7))

def test_fixed_batch_size_model_processes_all_inputs_as_single_batch(
self):
model_version = 1

def predict_func(inputs):
batch_size = len(inputs)
return [str(batch_size) for _ in inputs]

fixed_batch_size = 9
total_num_queries = fixed_batch_size
deploy_python_closure(
self.clipper_conn,
self.model_name_4,
model_version,
self.input_type,
predict_func,
batch_size=fixed_batch_size)
self.clipper_conn.link_model_to_app(self.app_name_4, self.model_name_4)
time.sleep(60)

addr = self.clipper_conn.get_query_addr()
url = "http://{addr}/{app}/predict".format(
addr=addr, app=self.app_name_4)
test_input = [[float(x) + (j * .001) for x in range(5)]
for j in range(total_num_queries)]
req_json = json.dumps({'input_batch': test_input})
headers = {'Content-type': 'application/json'}
response = requests.post(url, headers=headers, data=req_json)
parsed_response = response.json()
num_max_batch_queries = 0
for prediction in parsed_response["batch_predictions"]:
batch_size = prediction["output"]
if batch_size != self.default_output and int(
batch_size) == fixed_batch_size:
num_max_batch_queries += 1

self.assertEqual(num_max_batch_queries,
total_num_queries)

def test_remove_inactive_container(self):
container_name = "{}/noop-container:{}".format(clipper_registry,
clipper_version)
Expand Down Expand Up @@ -902,7 +941,8 @@ def test_remove_inactive_container(self):
'test_deployed_model_queried_successfully',
'test_batch_queries_returned_successfully',
'test_deployed_python_closure_queried_successfully',
'test_fixed_batch_size_model_processes_specified_query_batch_size_when_saturated'
'test_fixed_batch_size_model_processes_specified_query_batch_size_when_saturated',
'test_fixed_batch_size_model_processes_all_inputs_as_single_batch'
]

if __name__ == '__main__':
Expand Down
28 changes: 16 additions & 12 deletions src/benchmarks/src/end_to_end_bench.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,28 +80,32 @@ void send_predictions(std::unordered_map<std::string, std::string> &config,
query_data_raw[1] = thread_id;
}

std::shared_ptr<PredictionData> input = std::make_shared<DoubleVector>(
std::move(query_data), query_vec.size());
std::vector<std::shared_ptr<PredictionData>> input_batch;
input_batch.push_back(std::make_shared<DoubleVector>(
std::move(query_data), query_vec.size()));
Query q = {TEST_APPLICATION_LABEL,
UID,
input,
input_batch,
latency_objective,
clipper::DefaultOutputSelectionPolicy::get_name(),
{VersionedModelId(model_name, model_version)}};

folly::Future<Response> prediction = qp.predict(q);
folly::Future<std::vector<Response>> prediction = qp.predict(q);
bench_metrics.request_throughput_->mark(1);

std::move(prediction).thenValue([bench_metrics](Response r) {
std::move(prediction)
.thenValue([bench_metrics](std::vector<Response> responses) {
// Update metrics
if (r.output_is_default_) {
bench_metrics.default_pred_ratio_->increment(1, 1);
} else {
bench_metrics.default_pred_ratio_->increment(0, 1);
for (const auto &r : responses) {
if (r.output_is_default_) {
bench_metrics.default_pred_ratio_->increment(1, 1);
} else {
bench_metrics.default_pred_ratio_->increment(0, 1);
}
bench_metrics.latency_->insert(r.duration_micros_);
bench_metrics.num_predictions_->increment(1);
bench_metrics.throughput_->mark(1);
}
bench_metrics.latency_->insert(r.duration_micros_);
bench_metrics.num_predictions_->increment(1);
bench_metrics.throughput_->mark(1);
});
}

Expand Down
27 changes: 11 additions & 16 deletions src/frontends/src/query_frontend.hpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -346,17 +346,16 @@ class RequestHandler {
std::shared_ptr<HttpServer::Response> response,
std::shared_ptr<HttpServer::Request> request) {
try {
folly::Future<std::vector<folly::Try<Response>>> predictions =
folly::Future<std::vector<Response>> predictions =
decode_and_handle_predict(request->content.string(), name, policy,
latency_slo_micros, input_type);

std::move(predictions)
.thenValue([response,
app_metrics](std::vector<folly::Try<Response>> tries) {
app_metrics](std::vector<Response> responses) {
std::vector<std::string> all_content;
for (auto t : tries) {
for (const auto& r : responses) {
try {
Response r = t.value();
if (r.output_is_default_) {
app_metrics.default_pred_ratio_->increment(1, 1);
} else {
Expand Down Expand Up @@ -458,10 +457,10 @@ class RequestHandler {
}

static const std::string parse_output_y_hat(
std::shared_ptr<PredictionData>& y_hat) {
const std::shared_ptr<PredictionData>& y_hat) {
SharedPoolPtr<char> str_content = clipper::get_data<char>(y_hat);
return std::string(str_content.get() + y_hat->start(),
str_content.get() + y_hat->start() + y_hat->size());
return std::string(str_content.get() + y_hat->start_byte(),
str_content.get() + y_hat->start_byte() + y_hat->byte_size());
}

void delete_application(std::string name) {
Expand All @@ -483,7 +482,7 @@ class RequestHandler {
* }
*/
static const std::string get_prediction_response_content(
Response& query_response) {
const Response& query_response) {
rapidjson::Document json_response;
json_response.SetObject();
clipper::json::add_long(json_response, PREDICTION_RESPONSE_KEY_QUERY_ID,
Expand Down Expand Up @@ -570,7 +569,7 @@ class RequestHandler {
return clipper::json::to_json_string(error_response);
}

folly::Future<std::vector<folly::Try<Response>>> decode_and_handle_predict(
folly::Future<std::vector<Response>> decode_and_handle_predict(
std::string content, std::string name, std::string policy,
long latency_slo_micros, InputType input_type) {
rapidjson::Document d;
Expand Down Expand Up @@ -626,13 +625,9 @@ class RequestHandler {

std::vector<std::shared_ptr<PredictionData>> input_batch =
clipper::json::parse_inputs(input_type, d);
std::vector<folly::Future<Response>> predictions;
for (auto input : input_batch) {
auto prediction = query_processor_.predict(Query{
name, uid, input, latency_slo_micros, policy, versioned_models});
predictions.push_back(std::move(prediction));
}
return folly::collectAll(predictions);
auto predictions = query_processor_.predict(Query{
name, uid, input_batch, latency_slo_micros, policy, versioned_models});
return predictions;
}

/*
Expand Down
84 changes: 31 additions & 53 deletions src/frontends/src/query_frontend_tests.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,12 +20,17 @@ namespace {
class MockQueryProcessor {
public:
MockQueryProcessor() = default;
folly::Future<Response> predict(Query query) {
Response response(query, 3, 5, Output("-1.0", {VersionedModelId("m", "1")}),
false, boost::optional<std::string>{});
return folly::makeFuture(response);
folly::Future<std::vector<Response>> predict(const Query& query) {
std::vector<Response> responses;
responses.reserve(query.input_batch_.size());
for (const auto& input : query.input_batch_) {
responses.emplace_back(
3, 5, Output(input, {VersionedModelId("m", "1")}), false,
boost::optional<std::string>{});
}
return folly::makeFuture(responses);
}
folly::Future<FeedbackAck> update(FeedbackQuery /*feedback*/) {
folly::Future<FeedbackAck> update(const FeedbackQuery& /*feedback*/) {
return folly::makeFuture(true);
}

Expand Down Expand Up @@ -68,119 +73,99 @@ class QueryFrontendTest : public ::testing::Test {

TEST_F(QueryFrontendTest, TestDecodeCorrectInputInts) {
std::string test_json_ints = "{\"input\": [1,2,3,4]}";
std::vector<folly::Try<Response>> responses =
std::vector<Response> responses =
rh_.decode_and_handle_predict(test_json_ints, "test", "test_policy",
30000, InputType::Ints)
.get();
Response response = responses[0].value();
const Response& response = responses[0];

Query parsed_query = response.query_;
std::shared_ptr<IntVector> parsed_input =
std::dynamic_pointer_cast<IntVector>(parsed_query.input_);
std::dynamic_pointer_cast<IntVector>(response.output_.y_hat_);
int* data = get_data(parsed_input).get();
std::vector<int> parsed_input_data(
data + parsed_input->start(),
data + parsed_input->start() + parsed_input->size());

std::vector<int> expected_input_data{1, 2, 3, 4};
EXPECT_EQ(parsed_input_data, expected_input_data);
EXPECT_EQ(parsed_query.label_, "test");
EXPECT_EQ(parsed_query.latency_budget_micros_, 30000);
EXPECT_EQ(parsed_query.selection_policy_, "test_policy");
}

TEST_F(QueryFrontendTest, TestDecodeCorrectInputIntsBatch) {
std::string test_json_ints =
"{\"input_batch\": [[1, 2], [10, 20], [100, 200]]}";
std::vector<std::vector<int>> expected_input_data{
{1, 2}, {10, 20}, {100, 200}};
std::vector<folly::Try<Response>> responses =
std::vector<Response> responses =
rh_.decode_and_handle_predict(test_json_ints, "test", "test_policy",
30000, InputType::Ints)
.get();
for (size_t index = 0; index < responses.size(); ++index) {
Response response = responses[index].value();
Query parsed_query = response.query_;
const Response& response = responses[index];

std::shared_ptr<IntVector> parsed_input =
std::dynamic_pointer_cast<IntVector>(parsed_query.input_);
std::dynamic_pointer_cast<IntVector>(response.output_.y_hat_);
int* data = get_data(parsed_input).get();
std::vector<int> parsed_input_data(
data + parsed_input->start(),
data + parsed_input->start() + parsed_input->size());

EXPECT_EQ(parsed_input_data, expected_input_data[index]);
EXPECT_EQ(parsed_query.label_, "test");
EXPECT_EQ(parsed_query.latency_budget_micros_, 30000);
EXPECT_EQ(parsed_query.selection_policy_, "test_policy");
}
}

TEST_F(QueryFrontendTest, TestDecodeCorrectInputDoubles) {
std::string test_json_doubles = "{\"input\": [1.4,2.23,3.243242,0.3223424]}";
std::vector<folly::Try<Response>> responses =
std::vector<Response> responses =
rh_.decode_and_handle_predict(test_json_doubles, "test", "test_policy",
30000, InputType::Doubles)
.get();
Response response = responses[0].value();

Query parsed_query = response.query_;
const Response& response = responses[0];

std::shared_ptr<DoubleVector> parsed_input =
std::dynamic_pointer_cast<DoubleVector>(parsed_query.input_);
std::dynamic_pointer_cast<DoubleVector>(response.output_.y_hat_);
double* data = get_data(parsed_input).get();
std::vector<double> parsed_input_data(
data + parsed_input->start(),
data + parsed_input->start() + parsed_input->size());

std::vector<double> expected_input_data{1.4, 2.23, 3.243242, 0.3223424};
EXPECT_EQ(parsed_input_data, expected_input_data);
EXPECT_EQ(parsed_query.label_, "test");
EXPECT_EQ(parsed_query.latency_budget_micros_, 30000);
EXPECT_EQ(parsed_query.selection_policy_, "test_policy");
}

TEST_F(QueryFrontendTest, TestDecodeCorrectInputDoublesBatch) {
std::string test_json_doubles =
"{\"input_batch\": [[1.1, 2.2], [10.1, 20.2], [100.1, 200.2]]}";
std::vector<std::vector<double>> expected_input_data{
{1.1, 2.2}, {10.1, 20.2}, {100.1, 200.2}};
std::vector<folly::Try<Response>> responses =
std::vector<Response> responses =
rh_.decode_and_handle_predict(test_json_doubles, "test", "test_policy",
30000, InputType::Doubles)
.get();
for (size_t index = 0; index < responses.size(); ++index) {
Response response = responses[index].value();
Query parsed_query = response.query_;
const Response& response = responses[index];

std::shared_ptr<DoubleVector> parsed_input =
std::dynamic_pointer_cast<DoubleVector>(parsed_query.input_);
std::dynamic_pointer_cast<DoubleVector>(response.output_.y_hat_);
double* data = get_data(parsed_input).get();
std::vector<double> parsed_input_data(
data + parsed_input->start(),
data + parsed_input->start() + parsed_input->size());

EXPECT_EQ(parsed_input_data, expected_input_data[index]);
EXPECT_EQ(parsed_query.label_, "test");
EXPECT_EQ(parsed_query.latency_budget_micros_, 30000);
EXPECT_EQ(parsed_query.selection_policy_, "test_policy");
}
}

TEST_F(QueryFrontendTest, TestDecodeCorrectInputString) {
std::string test_json_string =
"{\"input\": \"hello world. This is a test string with "
"punctionation!@#$Y#;}#\"}";
std::vector<folly::Try<Response>> responses =
std::vector<Response> responses =
rh_.decode_and_handle_predict(test_json_string, "test", "test_policy",
30000, InputType::Strings)
.get();
Response response = responses[0].value();

Query parsed_query = response.query_;
const Response& response = responses[0];

std::shared_ptr<SerializableString> parsed_input =
std::dynamic_pointer_cast<SerializableString>(parsed_query.input_);
std::dynamic_pointer_cast<SerializableString>(response.output_.y_hat_);
char* data = get_data(parsed_input).get();
std::string parsed_input_data(
data + parsed_input->start(),
Expand All @@ -189,34 +174,27 @@ TEST_F(QueryFrontendTest, TestDecodeCorrectInputString) {
std::string expected_input_data(
"hello world. This is a test string with punctionation!@#$Y#;}#");
EXPECT_EQ(parsed_input_data, expected_input_data);
EXPECT_EQ(parsed_query.label_, "test");
EXPECT_EQ(parsed_query.latency_budget_micros_, 30000);
EXPECT_EQ(parsed_query.selection_policy_, "test_policy");
}

TEST_F(QueryFrontendTest, TestDecodeCorrectInputStringBatch) {
std::string test_json_strings =
"{\"input_batch\": [ \"this\", \"is\", \"a\", \"test\" ]}";
std::vector<std::string> expected_input_data{"this", "is", "a", "test"};
std::vector<folly::Try<Response>> responses =
std::vector<Response> responses =
rh_.decode_and_handle_predict(test_json_strings, "test", "test_policy",
30000, InputType::Strings)
.get();
for (size_t index = 0; index < responses.size(); ++index) {
Response response = responses[index].value();
Query parsed_query = response.query_;
const Response& response = responses[index];

std::shared_ptr<SerializableString> parsed_input =
std::dynamic_pointer_cast<SerializableString>(parsed_query.input_);
std::dynamic_pointer_cast<SerializableString>(response.output_.y_hat_);
char* data = get_data(parsed_input).get();
std::string parsed_input_data(
data + parsed_input->start(),
data + parsed_input->start() + parsed_input->size());

EXPECT_EQ(parsed_input_data, expected_input_data[index]);
EXPECT_EQ(parsed_query.label_, "test");
EXPECT_EQ(parsed_query.latency_budget_micros_, 30000);
EXPECT_EQ(parsed_query.selection_policy_, "test_policy");
}
}

Expand Down Expand Up @@ -339,11 +317,11 @@ TEST_F(QueryFrontendTest, TestDeleteManyApplications) {
TEST_F(QueryFrontendTest,
TestJsonResponseForSuccessfulPredictionFormattedCorrectly) {
std::string test_json = "{\"uid\": 1, \"input\": [1,2,3]}";
std::vector<folly::Try<Response>> responses =
std::vector<Response> responses =
rh_.decode_and_handle_predict(test_json, "test", "test_policy", 30000,
InputType::Ints)
.get();
Response response = responses[0].value();
const Response& response = responses[0];

std::string json_response = rh_.get_prediction_response_content(response);
rapidjson::Document parsed_response;
Expand All @@ -360,7 +338,7 @@ TEST_F(QueryFrontendTest,
->value.IsInt());
ASSERT_TRUE(parsed_response.GetObject()
.FindMember(PREDICTION_RESPONSE_KEY_OUTPUT)
->value.IsFloat());
->value.IsString());
ASSERT_TRUE(parsed_response.GetObject()
.FindMember(PREDICTION_RESPONSE_KEY_USED_DEFAULT)
->value.IsBool());
Expand Down
Loading