|
| 1 | +//***************************************************************************** |
| 2 | +// Copyright 2025 Intel Corporation |
| 3 | +// |
| 4 | +// Licensed under the Apache License, Version 2.0 (the "License"); |
| 5 | +// you may not use this file except in compliance with the License. |
| 6 | +// You may obtain a copy of the License at |
| 7 | +// |
| 8 | +// http://www.apache.org/licenses/LICENSE-2.0 |
| 9 | +// |
| 10 | +// Unless required by applicable law or agreed to in writing, software |
| 11 | +// distributed under the License is distributed on an "AS IS" BASIS, |
| 12 | +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| 13 | +// See the License for the specific language governing permissions and |
| 14 | +// limitations under the License. |
| 15 | +//***************************************************************************** |
| 16 | +#include "graphqueue.hpp" |
| 17 | + |
| 18 | +#include <atomic> |
| 19 | +#include <condition_variable> |
| 20 | +#include <future> |
| 21 | +#include <memory> |
| 22 | +#include <mutex> |
| 23 | +#include <optional> |
| 24 | +#include <queue> |
| 25 | +#include <thread> |
| 26 | +#include <utility> |
| 27 | +#include <vector> |
| 28 | + |
| 29 | +#include "../queue.hpp" |
| 30 | +#include "src/python/pythonnoderesources.hpp" |
| 31 | +#include "src/llm/servable.hpp" |
| 32 | + |
| 33 | +#include "mediapipe/framework/calculator_graph.h" |
| 34 | +#include "mediapipe/framework/port/status.h" |
| 35 | + |
| 36 | +#include "outputstreamobserver.hpp" |
| 37 | +namespace { |
| 38 | +//const ::mediapipe::Timestamp STARTING_TIMESTAMP = ::mediapipe::Timestamp(0); // TODO @atobisze common |
| 39 | +const std::string PYTHON_SESSION_SIDE_PACKET_NAME = "py"; |
| 40 | +const std::string LLM_SESSION_SIDE_PACKET_NAME = "llm"; |
| 41 | +} // namespace |
| 42 | +namespace ovms { |
| 43 | +GraphQueue::GraphQueue( const ::mediapipe::CalculatorGraphConfig& config, std::shared_ptr<PythonNodeResourcesMap> pythonNodeResourcesMap, std::shared_ptr<GenAiServableMap> genAiServableMap, int streamsLength) : |
| 44 | + Queue(streamsLength), |
| 45 | + pythonNodeResourcesMap(pythonNodeResourcesMap), |
| 46 | + genAiServableMap(genAiServableMap) { |
| 47 | + SPDLOG_ERROR("ER Constr graph queue:{}", (void*)this); |
| 48 | + inferRequests.reserve(streamsLength); |
| 49 | + // TODO FIXME split constructor to init to handle retCodes? |
| 50 | + for (auto i = 0; i < streamsLength; ++i) { |
| 51 | + auto gh = std::make_shared<GraphHelper>(); |
| 52 | + gh->graph = std::make_shared<::mediapipe::CalculatorGraph>(); |
| 53 | + gh->currentTimestamp = ::mediapipe::Timestamp(0); |
| 54 | + |
| 55 | + auto absStatus = gh->graph->Initialize(config); |
| 56 | + if (!absStatus.ok()) { |
| 57 | + SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this); |
| 58 | + throw 42; |
| 59 | + } |
| 60 | + for (auto& name : config.output_stream()) { |
| 61 | + std::string streamName = getStreamName(name); |
| 62 | + gh->outStreamObservers[streamName] = std::shared_ptr<OutputStreamObserverI>(new NullOutputStreamObserver()); // TODO use at() FIXME |
| 63 | + auto& perGraphObserverFunctor = gh->outStreamObservers[streamName]; |
| 64 | + absStatus = gh->graph->ObserveOutputStream(streamName, [&perGraphObserverFunctor](const ::mediapipe::Packet& packet) -> absl::Status { return perGraphObserverFunctor->handlePacket(packet); }); // TODO FIXME throw? |
| 65 | + if (!absStatus.ok()) { |
| 66 | + SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this); |
| 67 | + throw 42; |
| 68 | + } |
| 69 | + } |
| 70 | + std::map<std::string, mediapipe::Packet> inputSidePackets; |
| 71 | + inputSidePackets[PYTHON_SESSION_SIDE_PACKET_NAME] = mediapipe::MakePacket<PythonNodeResourcesMap>(*pythonNodeResourcesMap) |
| 72 | + .At(STARTING_TIMESTAMP); |
| 73 | + inputSidePackets[LLM_SESSION_SIDE_PACKET_NAME] = mediapipe::MakePacket<GenAiServableMap>(*genAiServableMap).At(STARTING_TIMESTAMP); |
| 74 | + for (auto [k, v] : inputSidePackets) { |
| 75 | + SPDLOG_ERROR("k:{} v", k); |
| 76 | + } |
| 77 | + SPDLOG_ERROR("ER"); |
| 78 | + absStatus = gh->graph->StartRun(inputSidePackets); |
| 79 | + SPDLOG_ERROR("ER"); |
| 80 | + if (!absStatus.ok()) { |
| 81 | + SPDLOG_ERROR("Input sidePackets size:{}, python map size:{} key:{} side packet name:{}", inputSidePackets.size(), pythonNodeResourcesMap->size(), pythonNodeResourcesMap->begin()->first, PYTHON_SESSION_SIDE_PACKET_NAME); |
| 82 | + SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this); |
| 83 | + throw 42; |
| 84 | + } |
| 85 | + |
| 86 | + SPDLOG_ERROR("ER"); |
| 87 | + inferRequests.emplace_back(std::move(gh)); |
| 88 | + SPDLOG_ERROR("ER"); |
| 89 | + } |
| 90 | +} |
| 91 | +GraphQueue::~GraphQueue() { |
| 92 | + SPDLOG_ERROR("ER Destroy graph queue:{}", (void*)this); |
| 93 | + for (auto& graphHelper : inferRequests) { |
| 94 | + auto absStatus = graphHelper->graph->WaitUntilIdle(); |
| 95 | + if (!absStatus.ok()) { |
| 96 | + SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this); |
| 97 | + // throw 42.2; |
| 98 | + } |
| 99 | + absStatus = graphHelper->graph->CloseAllPacketSources(); |
| 100 | + if (!absStatus.ok()) { |
| 101 | + SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this); |
| 102 | + // throw "as"; |
| 103 | + } |
| 104 | + absStatus = graphHelper->graph->WaitUntilDone(); |
| 105 | + if (!absStatus.ok()) { |
| 106 | + SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this); |
| 107 | + // throw 42.2; |
| 108 | + } |
| 109 | + graphHelper->graph->Cancel(); |
| 110 | + if (!absStatus.ok()) { |
| 111 | + SPDLOG_ERROR("ER issue:{} {}", absStatus.ToString(), (void*)this); |
| 112 | + // throw 42.2; |
| 113 | + } |
| 114 | + SPDLOG_ERROR("ER"); |
| 115 | + graphHelper->graph.reset(); |
| 116 | + SPDLOG_ERROR("ER"); |
| 117 | + } |
| 118 | + SPDLOG_ERROR("ER Destroy graph queue:{}", (void*)this); |
| 119 | +} |
| 120 | +} // namespace ovms |
0 commit comments