/*

* BSD 2-Clause License

*

* Copyright (c) 2021-2024, Hewlett Packard Enterprise

* All rights reserved.

*

* Redistribution and use in source and binary forms, with or without

* modification, are permitted provided that the following conditions are met:

*

* 1. Redistributions of source code must retain the above copyright notice, this

* list of conditions and the following disclaimer.

*

* 2. Redistributions in binary form must reproduce the above copyright notice,

* this list of conditions and the following disclaimer in the documentation

* and/or other materials provided with the distribution.

*

* THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"

* AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE

* IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE

* DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE

* FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL

* DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR

* SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER

* CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,

* OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE

* OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.

*/

#include <ctype.h>

#include <algorithm>

#include <cctype>

#include <stdlib.h>

#include <fcntl.h>

#include <sys/types.h>

#include <unistd.h>

#include "client.h"

#include "srexception.h"

#include "logger.h"

#include "utility.h"

#include "configoptions.h"

using namespace SmartRedis;

// Simple Client constructor

Client::Client(const char* logger_name)

: SRObject(logger_name)

{

// Create our ConfigOptions object (default: no suffixing)

auto cfgopts = ConfigOptions::create_from_environment("");

_cfgopts = cfgopts.release();

_cfgopts->_set_log_context(this);

// Log that a new client has been instantiated

log_data(LLDebug, "New client created");

// Establish our server connection

_establish_server_connection();

}

// Constructor with config options

Client::Client(ConfigOptions* cfgopts, const std::string& logger_name)

: SRObject(logger_name), _cfgopts(cfgopts->clone())

{

// Log that a new client has been instantiated

_cfgopts->_set_log_context(this);

log_data(LLDebug, "New client created");

// Establish our server connection

_establish_server_connection();

}

// Initialize a connection to the back-end database

void Client::_establish_server_connection()

{

// See what type of connection the user wants

std::string server_type = _cfgopts->_resolve_string_option(

"SR_DB_TYPE", "Clustered");

std::transform(server_type.begin(), server_type.end(), server_type.begin(),

[](unsigned char c){ return std::tolower(c); });

// Set up Redis server connection

// A std::bad_alloc exception on the initializer will be caught

// by the call to new for the client

if (server_type == "clustered") {

log_data(LLDeveloper, "Instantiating clustered Redis connection");

_redis_cluster = new RedisCluster(_cfgopts);

_redis = NULL;

_redis_server = _redis_cluster;

}

else { // Standalone or Colocated

log_data(LLDeveloper, "Instantiating standalone Redis connection");

_redis_cluster = NULL;

_redis = new Redis(_cfgopts);

_redis_server = _redis;

}

log_data(LLDeveloper, "Redis connection established");

// Initialize key prefixing

_get_prefix_settings();

_use_tensor_prefix = true;

_use_dataset_prefix = true;

_use_model_prefix = false;

_use_list_prefix = true;

}

// Constructor (deprecated)

Client::Client(bool cluster, const std::string& logger_name)

: SRObject(logger_name)

{

// Log that a new client has been instantiated

log_data(LLDebug, "New client created");

// Create our ConfigOptions object (default = no suffixing)

auto cfgopts = ConfigOptions::create_from_environment("");

_cfgopts = cfgopts.release();

_cfgopts->_set_log_context(this);

// Set up Redis server connection

// A std::bad_alloc exception on the initializer will be caught

// by the call to new for the client

_redis_cluster = (cluster ? new RedisCluster(_cfgopts) : NULL);

_redis = (cluster ? NULL : new Redis(_cfgopts));

if (cluster)

_redis_server = _redis_cluster;

else

_redis_server = _redis;

// Initialize key prefixing

_get_prefix_settings();

_use_tensor_prefix = true;

_use_dataset_prefix = true;

_use_model_prefix = false;

_use_list_prefix = true;

}

// Destructor

Client::~Client()

{

if (_redis_cluster != NULL)

{

delete _redis_cluster;

_redis_cluster = NULL;

}

if (_redis != NULL)

{

delete _redis;

_redis = NULL;

}

_redis_server = NULL;

delete _cfgopts;

_cfgopts = NULL;

// Log Client destruction

log_data(LLDebug, "Client destroyed");

}

// Put a DataSet object into the database

void Client::put_dataset(DataSet& dataset)

{

// Track calls to this API function

LOG_API_FUNCTION();

CommandList cmds;

_append_dataset_metadata_commands(cmds, dataset);

_append_dataset_tensor_commands(cmds, dataset);

_append_dataset_ack_command(cmds, dataset);

_redis_server->run_in_pipeline(cmds);

}

// Retrieve a DataSet object from the database

DataSet Client::get_dataset(const std::string& name)

{

// Track calls to this API function

LOG_API_FUNCTION();

// Get the metadata message and construct DataSet

CommandReply reply = _get_dataset_metadata(name);

// If the reply has no elements, it didn't exist

if (reply.n_elements() == 0) {

throw SRKeyException("The requested DataSet, \"" +

name + "\", does not exist.");

}

DataSet dataset(name);

_unpack_dataset_metadata(dataset, reply);

// Build the tensor keys

std::vector<std::string> tensor_names = dataset.get_tensor_names();

if (tensor_names.size() == 0)

return dataset; // If no tensors, we're done

std::vector<std::string> tensor_keys;

std::transform(

tensor_names.cbegin(),

tensor_names.cend(),

std::back_inserter(tensor_keys),

[this, name](std::string s){

return _build_dataset_tensor_key(name, s, true);

});

// Retrieve DataSet tensors

PipelineReply tensors = _redis_server->get_tensors(tensor_keys);

// Put them into the dataset

for (size_t i = 0; i < tensor_names.size(); i++) {

_add_dataset_tensor(dataset, tensor_names[i], tensors[i]);

}

return dataset;

}

// Rename the current dataset

void Client::rename_dataset(const std::string& old_name,

const std::string& new_name)

{

// Track calls to this API function

LOG_API_FUNCTION();

copy_dataset(old_name, new_name);

delete_dataset(old_name);

}

// Clone the dataset to a new name

void Client::copy_dataset(const std::string& src_name,

const std::string& dest_name)

{

// Track calls to this API function

LOG_API_FUNCTION();

// Get the metadata message and construct DataSet

CommandReply reply = _get_dataset_metadata(src_name);

if (reply.n_elements() == 0) {

throw SRKeyException("The requested DataSet " +

src_name + " does not exist.");

}

DataSet dataset(src_name);

_unpack_dataset_metadata(dataset, reply);

// Build tensor keys for cloning

std::vector<std::string> tensor_names = dataset.get_tensor_names();

std::vector<std::string> tensor_src_names =

_build_dataset_tensor_keys(src_name, tensor_names, true);

std::vector<std::string> tensor_dest_names =

_build_dataset_tensor_keys(dest_name, tensor_names, false);

// Clone tensors

_redis_server->copy_tensors(tensor_src_names, tensor_dest_names);

// Update the DataSet name to the destination name

// so we can reuse the object for placing metadata

// and ack commands

dataset.set_name(dest_name);

CommandList put_meta_cmds;

_append_dataset_metadata_commands(put_meta_cmds, dataset);

_append_dataset_ack_command(put_meta_cmds, dataset);

(void)_redis_server->run_in_pipeline(put_meta_cmds);

}

// Delete a DataSet from the database.

// All tensors and metdata in the DataSet will be deleted.

void Client::delete_dataset(const std::string& name)

{

// Track calls to this API function

LOG_API_FUNCTION();

CommandReply reply = _get_dataset_metadata(name);

if (reply.n_elements() == 0) {

throw SRRuntimeException("The requested DataSet " +

name + " does not exist.");

}

DataSet dataset(name);

_unpack_dataset_metadata(dataset, reply);

// Delete the metadata (which contains the ack key)

MultiKeyCommand cmd;

cmd << "DEL" << Keyfield(_build_dataset_meta_key(dataset.get_name(), true));

// Add in all the tensors to be deleted

std::vector<std::string> tensor_names = dataset.get_tensor_names();

std::vector<std::string> tensor_keys =

_build_dataset_tensor_keys(dataset.get_name(), tensor_names, true);

cmd.add_keys(tensor_keys);

// Run the command

reply = _run(cmd);

_report_reply_errors(reply, "An error was encountered when executing "\

"DataSet " + name + " deletion.");

}

// Put a tensor into the database

void Client::put_tensor(const std::string& name,

const void* data,

const std::vector<size_t>& dims,

const SRTensorType type,

const SRMemoryLayout mem_layout)

{

// Track calls to this API function

LOG_API_FUNCTION();

std::string key = _build_tensor_key(name, false);

TensorBase* tensor = NULL;

try {

switch (type) {

case SRTensorTypeDouble:

tensor = new Tensor<double>(key, data, dims, type, mem_layout);

break;

case SRTensorTypeFloat:

tensor = new Tensor<float>(key, data, dims, type, mem_layout);

break;

case SRTensorTypeInt64:

tensor = new Tensor<int64_t>(key, data, dims, type, mem_layout);

break;

case SRTensorTypeInt32:

tensor = new Tensor<int32_t>(key, data, dims, type, mem_layout);

break;

case SRTensorTypeInt16:

tensor = new Tensor<int16_t>(key, data, dims, type, mem_layout);

break;

case SRTensorTypeInt8:

tensor = new Tensor<int8_t>(key, data, dims, type, mem_layout);

break;

case SRTensorTypeUint16:

tensor = new Tensor<uint16_t>(key, data, dims, type, mem_layout);

break;

case SRTensorTypeUint8:

tensor = new Tensor<uint8_t>(key, data, dims, type, mem_layout);

break;

default:

throw SRTypeException("Invalid type for put_tensor");

}

}

catch (std::bad_alloc& e) {

throw SRBadAllocException("tensor");

}

// Send the tensor

CommandReply reply = _redis_server->put_tensor(*tensor);

// Cleanup

delete tensor;

tensor = NULL;

_report_reply_errors(reply, "put_tensor failed");

}

// Put bytes into the database

void Client::put_bytes(const std::string& name,

const void* bytes,

const size_t n_bytes)

{

// Track calls to this API function

LOG_API_FUNCTION();

std::string key = _build_bytes_key(name, false);

// Send the tensor

CommandReply reply = _redis_server->put_bytes(key, bytes, n_bytes);

if (reply.has_error())

throw SRRuntimeException("put_bytes failed");

}

// Get byte data and return to the user via modifying the

// supplied void pointer.

void Client::get_bytes(const std::string& name,

void*& data,

size_t& n_bytes)

{

std::string get_key = _build_bytes_key(name, true);

CommandReply reply = _redis_server->get_bytes(get_key);

if (reply.has_error())

throw SRRuntimeException("put_bytes failed");

n_bytes = reply.str_len();

data = malloc(n_bytes);

std::memcpy(data, (void*)(reply.str()), n_bytes);

}

// Get byte data and fill an already allocated array

// memory space that has the specified size.

void Client::unpack_bytes(const std::string& name,

void* data,

const size_t n_bytes,

size_t& n_used_bytes)

{

// Track calls to this API function

LOG_API_FUNCTION();

std::string get_key = _build_bytes_key(name, true);

CommandReply reply = _redis_server->get_bytes(get_key);

if (reply.has_error())

throw SRRuntimeException("put_bytes failed");

if (n_bytes < reply.str_len()) {

throw SRRuntimeException("Provided number of bytes of " +

std::to_string(n_bytes) + " " +

"smaller than retrieved data size of " +

std::to_string(reply.str_len()) + ".");

}

n_used_bytes = reply.str_len();

std::memcpy(data, reply.str(), reply.str_len());

}

// Delete bytes from the database

void Client::delete_bytes(const std::string& name)

{

// Track calls to this API function

LOG_API_FUNCTION();

std::string key = _build_bytes_key(name, true);

CommandReply reply = _redis_server->delete_bytes(key);

if (reply.has_error())

throw SRRuntimeException("put_bytes failed");

}

// Get the tensor data, dimensions, and type for the provided tensor name.

// This function will allocate and retain management of the memory for the

// tensor data.

void Client::get_tensor(const std::string& name,

void*& data,

std::vector<size_t>& dims,

SRTensorType& type,

const SRMemoryLayout mem_layout)

{

// Track calls to this API function

LOG_API_FUNCTION();

// Retrieve the TensorBase from the database

TensorBase* ptr = _get_tensorbase_obj(name);

// Set the user values

dims = ptr->dims();

type = ptr->type();

data = ptr->data_view(mem_layout);

// Hold the Tensor in memory for memory management

_tensor_memory.add_tensor(ptr);

}

// Get the tensor data, dimensions, and type for the provided tensor name.

// This function will allocate and retain management of the memory for the

// tensor data and dimensions. This is a c-style interface for the tensor

// dimensions. Another function exists for std::vector dimensions.

void Client::get_tensor(const std::string& name,

void*& data,

size_t*& dims,

size_t& n_dims,

SRTensorType& type,

const SRMemoryLayout mem_layout)

{

// Track calls to this API function

LOG_API_FUNCTION();

std::vector<size_t> dims_vec;

get_tensor(name, data, dims_vec, type, mem_layout);

size_t dims_bytes = sizeof(size_t) * dims_vec.size();

dims = _dim_queries.allocate_bytes(dims_bytes);

n_dims = dims_vec.size();

std::vector<size_t>::const_iterator it = dims_vec.cbegin();

for (size_t i = 0; it != dims_vec.cend(); i++, it++)

dims[i] = *it;

}

// Get tensor data and fill an already allocated array memory space that

// has the specified MemoryLayout. The provided type and dimensions are

// checked against retrieved values to ensure the provided memory space is

// sufficient. This method is the most memory efficient way to retrieve

// tensor data.

void Client::unpack_tensor(const std::string& name,

void* data,

const std::vector<size_t>& dims,

const SRTensorType type,

const SRMemoryLayout mem_layout)

{

// Track calls to this API function

LOG_API_FUNCTION();

if (mem_layout == SRMemLayoutContiguous && dims.size() > 1) {

throw SRRuntimeException("The destination memory space "\

"dimension vector should only "\

"be of size one if the memory "\

"layout is contiguous.");

}

std::string get_key = _build_tensor_key(name, true);

CommandReply reply = _redis_server->get_tensor(get_key);

std::vector<size_t> reply_dims = GetTensorCommand::get_dims(reply);

// Make sure we have the right dims to unpack into (Contiguous case)

if (mem_layout == SRMemLayoutContiguous ||

mem_layout == SRMemLayoutFortranContiguous) {

size_t total_dims = 1;

for (size_t i = 0; i < reply_dims.size(); i++) {

total_dims *= reply_dims[i];

}

if (total_dims != dims[0] &&

mem_layout == SRMemLayoutContiguous) {

throw SRRuntimeException("The dimensions of the fetched "\

"tensor do not match the length of "\

"the contiguous memory space.");

}

}

// Make sure we have the right dims to unpack into (Nested case)

if (mem_layout == SRMemLayoutNested) {

if (dims.size() != reply_dims.size()) {

// Same number of dimensions

throw SRRuntimeException("The number of dimensions of the "\

"fetched tensor, " +

std::to_string(reply_dims.size()) + " "\

"does not match the number of "\

"dimensions of the user memory space, " +

std::to_string(dims.size()));

}

// Same size in each dimension

for (size_t i = 0; i < reply_dims.size(); i++) {

if (dims[i] != reply_dims[i]) {

throw SRRuntimeException("The dimensions of the fetched tensor "\

"do not match the provided "\

"dimensions of the user memory space.");

}

}

}

// Make sure we're unpacking the right type of data

SRTensorType reply_type = GetTensorCommand::get_data_type(reply);

if (type != reply_type)

throw SRRuntimeException("The type of the fetched tensor "\

"does not match the provided type");

// Retrieve the tensor data into a Tensor

std::string_view blob = GetTensorCommand::get_data_blob(reply);

TensorBase* tensor = NULL;

try {

switch (reply_type) {

case SRTensorTypeDouble:

tensor = new Tensor<double>(get_key, (void*)blob.data(),

reply_dims, reply_type,

SRMemLayoutContiguous);

break;

case SRTensorTypeFloat:

tensor = new Tensor<float>(get_key, (void*)blob.data(),

reply_dims, reply_type,

SRMemLayoutContiguous);

break;

case SRTensorTypeInt64:

tensor = new Tensor<int64_t>(get_key, (void*)blob.data(),

reply_dims, reply_type,

SRMemLayoutContiguous);

break;

case SRTensorTypeInt32:

tensor = new Tensor<int32_t>(get_key, (void*)blob.data(),

reply_dims, reply_type,

SRMemLayoutContiguous);

break;

case SRTensorTypeInt16:

tensor = new Tensor<int16_t>(get_key, (void*)blob.data(),

reply_dims, reply_type,

SRMemLayoutContiguous);

break;

case SRTensorTypeInt8:

tensor = new Tensor<int8_t>(get_key, (void*)blob.data(),

reply_dims, reply_type,

SRMemLayoutContiguous);

break;

case SRTensorTypeUint16:

tensor = new Tensor<uint16_t>(get_key, (void*)blob.data(),

reply_dims, reply_type,

SRMemLayoutContiguous);

break;

case SRTensorTypeUint8:

tensor = new Tensor<uint8_t>(get_key, (void*)blob.data(),

reply_dims, reply_type,

SRMemLayoutContiguous);

break;

default:

throw SRTypeException("Invalid type for unpack_tensor");

}

}

catch (std::bad_alloc& e) {

throw SRBadAllocException("tensor");

}

// Unpack the tensor and reclaim it

tensor->fill_mem_space(data, dims, mem_layout);

delete tensor;

tensor = NULL;

}

// Move a tensor from one name to another name

void Client::rename_tensor(const std::string& old_name,

const std::string& new_name)

{

// Track calls to this API function

LOG_API_FUNCTION();

std::string old_key = _build_tensor_key(old_name, true);

std::string new_key = _build_tensor_key(new_name, false);

CommandReply reply = _redis_server->rename_tensor(old_key, new_key);

_report_reply_errors(reply, "rename_tensor failed");

}

// Delete a tensor from the database

void Client::delete_tensor(const std::string& name)

{

// Track calls to this API function

LOG_API_FUNCTION();

std::string key = _build_tensor_key(name, true);

CommandReply reply = _redis_server->delete_tensor(key);

_report_reply_errors(reply, "delete_tensor failed");

}

// Copy the tensor from the source name to the destination name

void Client::copy_tensor(const std::string& src_name,

const std::string& dest_name)

{

// Track calls to this API function

LOG_API_FUNCTION();

std::string src_key = _build_tensor_key(src_name, true);

std::string dest_key = _build_tensor_key(dest_name, false);

CommandReply reply = _redis_server->copy_tensor(src_key, dest_key);

_report_reply_errors(reply, "copy_tensor failed");

}

// Set a model from file in the database for future execution

void Client::set_model_from_file(const std::string& name,

const std::string& model_file,

const std::string& backend,

const std::string& device,

int batch_size,

int min_batch_size,

int min_batch_timeout,

const std::string& tag,

const std::vector<std::string>& inputs,

const std::vector<std::string>& outputs)

{

// Track calls to this API function

LOG_API_FUNCTION();

if (model_file.size() == 0) {

throw SRParameterException("model_file is a required "

"parameter of set_model_from_file.");

}

std::ifstream fin(model_file, std::ios::binary);

std::ostringstream ostream;

ostream << fin.rdbuf();

const std::string tmp = ostream.str();

std::string_view model(tmp.data(), tmp.length());

set_model(name, model, backend, device, batch_size,

min_batch_size, min_batch_timeout, tag, inputs, outputs);

}

// Set a model from file in the database for future execution in a multi-GPU system

void Client::set_model_from_file_multigpu(const std::string& name,

const std::string& model_file,

const std::string& backend,

int first_gpu,

int num_gpus,

int batch_size,

int min_batch_size,

int min_batch_timeout,

const std::string& tag,

const std::vector<std::string>& inputs,

const std::vector<std::string>& outputs)

{

// Track calls to this API function

LOG_API_FUNCTION();

if (model_file.size() == 0) {

throw SRParameterException("model_file is a required "

"parameter of set_model_from_file_multigpu.");

}

std::ifstream fin(model_file, std::ios::binary);

std::ostringstream ostream;

ostream << fin.rdbuf();

const std::string tmp = ostream.str();

std::string_view model(tmp.data(), tmp.length());

set_model_multigpu(name, model, backend, first_gpu, num_gpus, batch_size,

min_batch_size, min_batch_timeout, tag, inputs, outputs);

}

// Validate batch settings for the set_model calls

inline void __check_batch_settings(

int batch_size, int min_batch_size, int min_batch_timeout)

{

// Throw a usage exception if batch_size is zero but one of the other

// parameters is non-zero

if (batch_size == 0 && (min_batch_size > 0 || min_batch_timeout > 0)) {

throw SRRuntimeException(

"batch_size must be non-zero if min_batch_size or "

"min_batch_timeout is used; otherwise batching will "

"not be performed."

);

}

// Throw a usage exception if min_batch_timeout is nonzero and

// min_batch_size is zero. (batch_size also has to be non-zero, but

// this was caught in the previous clause.)

if (min_batch_timeout > 0 && min_batch_size == 0) {

throw SRRuntimeException(

"min_batch_size must be non-zero if min_batch_timeout "

"is used; otherwise the min_batch_timeout parameter is ignored."

);

}

// Issue a warning if min_batch_size is non-zero but min_batch_timeout is zero

if (min_batch_size > 0 && min_batch_timeout == 0) {

std::cerr << "WARNING: min_batch_timeout was not set when a non-zero "

<< "min_batch_size was selected. " << std::endl

<< "Setting a small value (~10ms) for min_batch_timeout "

<< "may improve performance" << std::endl;

}

}

// Set a model from a string buffer in the database for future execution

void Client::set_model(const std::string& name,

const std::string_view& model,

const std::string& backend,

const std::string& device,

int batch_size,

int min_batch_size,

int min_batch_timeout,

const std::string& tag,

const std::vector<std::string>& inputs,

const std::vector<std::string>& outputs)

{

// Track calls to this API function

LOG_API_FUNCTION();

if (name.size() == 0) {

throw SRParameterException("name is a required parameter of set_model.");

}

if (backend.size() == 0) {

throw SRParameterException("backend is a required "\

"parameter of set_model.");

}

if (backend.compare("TF") != 0) {

if (inputs.size() > 0) {

throw SRParameterException("INPUTS in the model set command "\

"is only valid for TF models");

}

if (outputs.size() > 0) {

throw SRParameterException("OUTPUTS in the model set command "\

"is only valid for TF models");

}

}

const char* backends[] = { "TF", "TFLITE", "TORCH", "ONNX" };

bool found = false;

for (size_t i = 0; i < sizeof(backends)/sizeof(backends[0]); i++)

found = found || (backend.compare(backends[i]) != 0);

if (!found) {

throw SRParameterException(backend + " is not a valid backend.");

}

if (device.size() == 0) {

throw SRParameterException("device is a required "

"parameter of set_model.");

}

if (device.compare("CPU") != 0 &&

std::string(device).find("GPU") == std::string::npos) {

throw SRRuntimeException(device + " is not a valid device.");

}

__check_batch_settings(batch_size, min_batch_size, min_batch_timeout);

// Split model into chunks

size_t offset = 0;

std::vector<std::string_view> model_segments;

size_t chunk_size = _redis_server->get_model_chunk_size();

size_t remaining = model.length();

for (offset = 0; offset < model.length(); offset += chunk_size) {

size_t this_chunk_size = remaining > chunk_size ? chunk_size : remaining;

std::string_view chunk(model.data() + offset, this_chunk_size);

model_segments.push_back(chunk);

remaining -= this_chunk_size;

}

std::string key = _build_model_key(name, false);

auto response = _redis_server->set_model(

key, model_segments, backend, device,

batch_size, min_batch_size, min_batch_timeout,

tag, inputs, outputs);

if (response.has_error()) {

throw SRInternalException(

"An unknown error occurred while setting the model");

}

}

void Client::set_model_multigpu(const std::string& name,

const std::string_view& model,

const std::string& backend,

int first_gpu,

int num_gpus,

int batch_size,

int min_batch_size,

int min_batch_timeout,

const std::string& tag,

const std::vector<std::string>& inputs,

const std::vector<std::string>& outputs)

{

// Track calls to this API function

LOG_API_FUNCTION();

if (name.size() == 0) {

throw SRParameterException("name is a required parameter of set_model.");

}

if (backend.size() == 0) {

throw SRParameterException("backend is a required "\

"parameter of set_model.");

}

if (backend.compare("TF") != 0) {

if (inputs.size() > 0) {

throw SRParameterException("INPUTS in the model set command "\

"is only valid for TF models");

}

if (outputs.size() > 0) {

throw SRParameterException("OUTPUTS in the model set command "\

"is only valid for TF models");

}

}

if (first_gpu < 0) {

throw SRParameterException("first_gpu must be a non-negative integer");

}

if (num_gpus < 1) {

throw SRParameterException("num_gpus must be a positive integer.");

}

const char* backends[] = { "TF", "TFLITE", "TORCH", "ONNX" };

bool found = false;

for (size_t i = 0; i < sizeof(backends)/sizeof(backends[0]); i++)

found = found || (backend.compare(backends[i]) != 0);

if (!found) {

throw SRParameterException(backend + " is not a valid backend.");

}

__check_batch_settings(batch_size, min_batch_size, min_batch_timeout);

// Split model into chunks

size_t offset = 0;

std::vector<std::string_view> model_segments;

size_t chunk_size = _redis_server->get_model_chunk_size();

size_t remaining = model.length();

for (offset = 0; offset < model.length(); offset += chunk_size) {

size_t this_chunk_size = remaining > chunk_size ? chunk_size : remaining;

std::string_view chunk(model.data() + offset, this_chunk_size);

model_segments.push_back(chunk);

remaining -= this_chunk_size;

}

std::string key = _build_model_key(name, false);

_redis_server->set_model_multigpu(

key, model_segments, backend, first_gpu, num_gpus,

batch_size, min_batch_size, min_batch_timeout,

tag, inputs, outputs);

}

// Retrieve the model from the database

std::string_view Client::get_model(const std::string& name)

{

// Track calls to this API function

LOG_API_FUNCTION();

// Get the model from the server

std::string get_key = _build_model_key(name, true);

CommandReply reply = _redis_server->get_model(get_key);

_report_reply_errors(reply, "failed to get model from server");

// In most cases, the reply will be a single string

// consisting of the serialized model

if (!reply.is_array()) {

char* model = _model_queries.allocate(reply.str_len());

if (model == NULL)

throw SRBadAllocException("model query");

std::memcpy(model, reply.str(), reply.str_len());

return std::string_view(model, reply.str_len());

}

// Otherwise, we need to concatenate the segments together

size_t model_length = 0;

size_t offset = 0;

for (size_t i = 0; i < reply.n_elements(); i++) {

model_length += reply[i].str_len();

}

char* model = _model_queries.allocate(model_length);

if (model == NULL)

throw SRBadAllocException("model query");

for (size_t i = 0; i < reply.n_elements(); i++) {

std::memcpy(model + offset, reply[i].str(), reply[i].str_len());

}

return std::string_view(model, model_length);

}

// Set a script from file in the database for future execution

void Client::set_script_from_file(const std::string& name,

const std::string& device,

const std::string& script_file)

{

// Track calls to this API function

LOG_API_FUNCTION();

// Read the script from the file

std::ifstream fin(script_file);

std::ostringstream ostream;

ostream << fin.rdbuf();

const std::string tmp = ostream.str();

std::string_view script(tmp.data(), tmp.length());

// Send it to the database

set_script(name, device, script);

}

// Set a script from file in the database for future execution

// in a multi-GPU system

void Client::set_script_from_file_multigpu(const std::string& name,

const std::string& script_file,

int first_gpu,

int num_gpus)

{

// Track calls to this API function

LOG_API_FUNCTION();

// Read the script from the file

std::ifstream fin(script_file);

std::ostringstream ostream;

ostream << fin.rdbuf();

const std::string tmp = ostream.str();

std::string_view script(tmp.data(), tmp.length());

// Send it to the database

set_script_multigpu(name, script, first_gpu, num_gpus);

}

// Set a script from a string buffer in the database for future execution

void Client::set_script(const std::string& name,

const std::string& device,

const std::string_view& script)

{

// Track calls to this API function

LOG_API_FUNCTION();

if (device.size() == 0) {

throw SRParameterException("device is a required "

"parameter of set_script.");

}

if (device.compare("CPU") != 0 &&

std::string(device).find("GPU") == std::string::npos) {

throw SRRuntimeException(device + " is not a valid device.");

}

std::string key = _build_model_key(name, false);

auto response = _redis_server->set_script(key, device, script);

if (response.has_error()) {

throw SRInternalException(

"An unknown error occurred while setting the script");

}

}

// Set a script in the database for future execution in a multi-GPU system

void Client::set_script_multigpu(const std::string& name,

const std::string_view& script,

int first_gpu,

int num_gpus)

{

// Track calls to this API function

LOG_API_FUNCTION();

if (first_gpu < 0) {

throw SRParameterException("first_gpu must be a non-negative integer.");

}

if (num_gpus < 1) {

throw SRParameterException("num_gpus must be a positive integer.");

}

std::string key = _build_model_key(name, false);

_redis_server->set_script_multigpu(key, script, first_gpu, num_gpus);