AthInfer::TritonTool::Impl Struct Reference

Inheritance diagram for AthInfer::TritonTool::Impl:

Collaboration diagram for AthInfer::TritonTool::Impl:

Public Member Functions
StatusCode	getClient (tc::InferenceServerGrpcClient *&client, const std::string &url, int port, bool useSSL) const
tc::Error	checkServerHealth (tc::InferenceServerGrpcClient &client) const
void	waitBeforeRetry (const int retryDelayMs) const
StatusCode	runInference (tc::InferenceServerGrpcClient &client, const std::vector< tc::InferInput * > &rawInputs, const int maxRetries, const int retryDelayMs, std::shared_ptr< tc::InferResult > &results) const
template<typename T>
StatusCode	prepareInput (const std::string &name, const std::vector< int64_t > &shape, const std::vector< T > &data, std::vector< std::unique_ptr< tc::InferInput > > &inputs) const
template<typename T>
StatusCode	extractOutput (const std::string &name, const tc::InferResult &result, std::vector< T > &outputVec) const
	AthMessaging (IMessageSvc *msgSvc, const std::string &name)
	Constructor.
	AthMessaging (const std::string &name)
	Constructor with auto-retrieval of the MessageSvc.
bool	msgLvl (const MSG::Level lvl) const
	Test the output level.
MsgStream &	msg () const
	The standard message stream.
MsgStream &	msg (const MSG::Level lvl) const
	The standard message stream.
void	setLevel (MSG::Level lvl)
	Change the current logging level.

Public Attributes
const AthAsynchronousAlgorithm *	m_parentAsyncAlg = nullptr
std::unique_ptr< tc::InferOptions >	m_options

Private Member Functions
void	initMessaging () const
	Initialize our message level and MessageSvc.

Private Attributes
std::string	m_nm
	Message source name.
boost::thread_specific_ptr< MsgStream >	m_msg_tls
	MsgStream instance (a std::cout like with print-out levels).
std::atomic< IMessageSvc * >	m_imsg { nullptr }
	MessageSvc pointer.
std::atomic< MSG::Level >	m_lvl { MSG::NIL }
	Current logging level.
std::atomic_flag m_initialized	ATLAS_THREAD_SAFE = ATOMIC_FLAG_INIT
	Messaging initialized (initMessaging).

Detailed Description

Definition at line 51 of file TritonTool.cxx.

Member Function Documentation

AthMessaging() [1/2]

AthMessaging::AthMessaging

(

const std::string &

name

)

Constructor with auto-retrieval of the MessageSvc.

Parameters

name	Name of the message stream

Definition at line 71 of file AthMessaging.cxx.

                                                 :
  m_nm(name)
{}

AthMessaging() [2/2]

AthMessaging::AthMessaging	(	IMessageSvc *	msgSvc,
		const std::string &	name )

Constructor.

Parameters

msgSvc	Pointer to the MessageSvc
name	Name of the message stream

Definition at line 66 of file AthMessaging.cxx.

                                                   :
  m_nm(name), m_imsg(msgSvc)
{}

checkServerHealth()

tc::Error AthInfer::TritonTool::Impl::checkServerHealth ( tc::InferenceServerGrpcClient & client ) const

inline

Definition at line 76 of file TritonTool.cxx.

                                                                       {
 
    tc::Headers httpHeaders;
    bool live = false;
    tc::Error err =
        client.IsServerLive(&live, httpHeaders, m_options->client_timeout_);
    if (!err.IsOk()) {
      return err;
    }
    if (!live) {
      return tc::Error("Triton server is not live");
    }
 
    bool serverReady = false;
    err = client.IsServerReady(&serverReady, httpHeaders,
                               m_options->client_timeout_);
    if (!err.IsOk()) {
      return err;
    }
    if (!serverReady) {
      return tc::Error("Triton server is not ready");
    }
 
    bool modelReady = false;
    err = client.IsModelReady(&modelReady, m_options->model_name_,
                              m_options->model_version_, httpHeaders,
                              m_options->client_timeout_);
    if (!err.IsOk()) {
      return err;
    }
    if (!modelReady) {
      return tc::Error("Triton model " + m_options->model_name_ + " is not ready");
    }
 
    return tc::Error::Success;
  }

extractOutput()

template<typename T>

StatusCode AthInfer::TritonTool::Impl::extractOutput ( const std::string & name, const tc::InferResult & result, std::vector< T > & outputVec ) const

inline

Definition at line 212 of file TritonTool.cxx.

                                                          {
 
    const uint8_t* rawData = nullptr;
    size_t size = 0;
 
    // Get access to the buffer holding raw results of specified output returned
    // by the server. Note: the buffer is owned by InferResult instance. Users
    // can copy out the data if required to extend the lifetime.
    TRITON_CHECK(result.RawData(name, &rawData, &size));
 
    outputVec.resize(size / sizeof(T));
    std::memcpy(outputVec.data(), rawData, size);
    return StatusCode::SUCCESS;
  }

getClient()

StatusCode AthInfer::TritonTool::Impl::getClient ( tc::InferenceServerGrpcClient *& client, const std::string & url, int port, bool useSSL ) const

inline

Definition at line 56 of file TritonTool.cxx.

                                                                          {
 
    thread_local std::unique_ptr<tc::InferenceServerGrpcClient> threadClient;
    if (!threadClient) {
 
      const std::string urlAndPort =
          url + ":" + std::to_string(port);  // always use the gRPC port
 
      constexpr bool verbose = false;
      TRITON_CHECK(tc::InferenceServerGrpcClient::Create(
          &threadClient, urlAndPort, verbose, useSSL));
 
      ATH_MSG_INFO("Triton client created for url: " << urlAndPort);
    }
    client = threadClient.get();
 
    return StatusCode::SUCCESS;
  }

initMessaging()

void AthMessaging::initMessaging ( ) const

privateinherited

Initialize our message level and MessageSvc.

This method should only be called once.

Definition at line 39 of file AthMessaging.cxx.

{
  m_imsg = Athena::getMessageSvc();
  // If user did not set an explicit level, set a default
  if (m_lvl == MSG::NIL) {
    m_lvl = m_imsg ?
      static_cast<MSG::Level>( m_imsg.load()->outputLevel(m_nm) ) :
      MSG::INFO;
  }
}

msg() [1/2]

MsgStream & AthMessaging::msg ( ) const

inlineinherited

The standard message stream.

Returns a reference to the default message stream May not be invoked before sysInitialize() has been invoked.

Definition at line 167 of file AthMessaging.h.

{
  MsgStream* ms = m_msg_tls.get();
  if (!ms) {
    if (!m_initialized.test_and_set()) initMessaging();
    ms = new MsgStream(m_imsg,m_nm);
    m_msg_tls.reset( ms );
  }
 
  ms->setLevel (m_lvl);
  return *ms;
}

msg() [2/2]

MsgStream & AthMessaging::msg ( const MSG::Level lvl ) const

inlineinherited

The standard message stream.

Returns a reference to the default message stream May not be invoked before sysInitialize() has been invoked.

Definition at line 182 of file AthMessaging.h.

{ return msg() << lvl; }

msgLvl()

bool AthMessaging::msgLvl ( const MSG::Level lvl ) const

inlineinherited

Test the output level.

Parameters

lvl	The message level to test against

Returns

boolean Indicating if messages at given level will be printed

Return values

true	Messages at level "lvl" will be printed

Definition at line 151 of file AthMessaging.h.

{
  // If user did not set explicit message level we have to initialize
  // the messaging and retrieve the default via the MessageSvc.
  if (m_lvl==MSG::NIL && !m_initialized.test_and_set()) initMessaging();
 
  if (m_lvl <= lvl) {
    msg() << lvl;
    return true;
  } else {
    return false;
  }
}

prepareInput()

template<typename T>

StatusCode AthInfer::TritonTool::Impl::prepareInput ( const std::string & name, const std::vector< int64_t > & shape, const std::vector< T > & data, std::vector< std::unique_ptr< tc::InferInput > > & inputs ) const

inline

Definition at line 184 of file TritonTool.cxx.

                                                            {
 
    const char* dtype = TritonDType<T>::value;
    tc::InferInput* rawInputPtr = nullptr;
 
    // create the InferInput object with the predefined name, shape, and data
    // type.
    TRITON_CHECK(tc::InferInput::Create(&rawInputPtr, name, shape, dtype));
    assert(rawInputPtr != nullptr);
 
    // Append tensor values for this input from a byte array.
    // Note: The vector is not copied and so it must not be modified or
    // destroyed until this input is no longer needed (that is until the Infer()
    // call(s) that use the input have completed). Multiple calls can be made to
    // this API to keep adding tensor data for this input. The data will be
    // delivered in the order it was added.
    std::unique_ptr<tc::InferInput> input{rawInputPtr};
    TRITON_CHECK(input->AppendRaw(reinterpret_cast<const uint8_t*>(data.data()),
                                  data.size() * sizeof(T)));
 
    inputs.push_back(std::move(input));
    return StatusCode::SUCCESS;
  }

runInference()

StatusCode AthInfer::TritonTool::Impl::runInference ( tc::InferenceServerGrpcClient & client, const std::vector< tc::InferInput * > & rawInputs, const int maxRetries, const int retryDelayMs, std::shared_ptr< tc::InferResult > & results ) const

inline

Definition at line 119 of file TritonTool.cxx.

                                                   {
 
    tc::Headers httpHeaders;
    grpc_compression_algorithm compressionAlgorithm =
        grpc_compression_algorithm::GRPC_COMPRESS_NONE;
 
    tc::Error err;
    for (int attempt = 0; attempt <= maxRetries; ++attempt) {
      if (m_parentAsyncAlg == nullptr) {
        tc::InferResult* rawResultPtr = nullptr;
        err = client.Infer(&rawResultPtr, *m_options, rawInputs, {},
                            httpHeaders, compressionAlgorithm);
        if (err.IsOk() && rawResultPtr != nullptr) {
          results.reset(rawResultPtr);
          err = results->RequestStatus();
        } else if (err.IsOk()) {
          err = tc::Error("Triton synchronous inference returned no result");
        }
      } else {
        using Promise_t = boost::fibers::promise<tc::InferResult*>;
        using Future_t = boost::fibers::future<tc::InferResult*>;
        Promise_t promise{};
        Future_t future = promise.get_future();
        auto callback = [&promise](tc::InferResult* resultPtr) {
          promise.set_value(resultPtr);
        };
        err = client.AsyncInfer(callback, *m_options, rawInputs, {},
                                httpHeaders, compressionAlgorithm);
        if (err.IsOk()) {
          results.reset(future.get());
          ATH_CHECK(m_parentAsyncAlg->restoreAfterSuspend());
          if (results != nullptr) {
            err = results->RequestStatus();
          } else {
            err = tc::Error("Triton asynchronous inference returned no "
                            "result");
          }
        }
      }
 
      if (err.IsOk()) {
        return StatusCode::SUCCESS;
      }
 
      if (attempt == maxRetries) {
        ATH_MSG_ERROR("Triton inference failed after " << (attempt + 1)
                                                       << " attempt(s): "
                                                       << err);
        return StatusCode::FAILURE;
      }
 
      ATH_MSG_WARNING("Triton inference attempt " << (attempt + 1)
                                                  << " failed: " << err
                                                  << "; retrying");
      waitBeforeRetry(retryDelayMs);
    }
 
    return StatusCode::FAILURE;
  }

setLevel()

void AthMessaging::setLevel ( MSG::Level lvl )

inherited

Change the current logging level.

Use this rather than msg().setLevel() for proper operation with MT.

Definition at line 28 of file AthMessaging.cxx.

{
  m_lvl = lvl;
}

waitBeforeRetry()

void AthInfer::TritonTool::Impl::waitBeforeRetry ( const int retryDelayMs ) const

inline

Definition at line 113 of file TritonTool.cxx.

                                                     {
    if (retryDelayMs > 0) {
      std::this_thread::sleep_for(std::chrono::milliseconds(retryDelayMs));
    }
  }

Member Data Documentation

ATLAS_THREAD_SAFE

std::atomic_flag m_initialized AthMessaging::ATLAS_THREAD_SAFE = ATOMIC_FLAG_INIT

mutableprivateinherited

Messaging initialized (initMessaging).

Definition at line 141 of file AthMessaging.h.

m_imsg

std::atomic<IMessageSvc*> AthMessaging::m_imsg { nullptr }

mutableprivateinherited

MessageSvc pointer.

Definition at line 135 of file AthMessaging.h.

{ nullptr };

m_lvl

std::atomic<MSG::Level> AthMessaging::m_lvl { MSG::NIL }

mutableprivateinherited

Current logging level.

Definition at line 138 of file AthMessaging.h.

{ MSG::NIL };

m_msg_tls

boost::thread_specific_ptr<MsgStream> AthMessaging::m_msg_tls

mutableprivateinherited

MsgStream instance (a std::cout like with print-out levels).

Definition at line 132 of file AthMessaging.h.

m_nm

std::string AthMessaging::m_nm

privateinherited

Message source name.

Definition at line 129 of file AthMessaging.h.

m_options

std::unique_ptr<tc::InferOptions> AthInfer::TritonTool::Impl::m_options

Definition at line 230 of file TritonTool.cxx.

m_parentAsyncAlg

const AthAsynchronousAlgorithm* AthInfer::TritonTool::Impl::m_parentAsyncAlg = nullptr

Definition at line 229 of file TritonTool.cxx.

---

The documentation for this struct was generated from the following file:

• TritonTool.cxx