Skip to content
Open
75 changes: 63 additions & 12 deletions src/pull_module/curl_downloader.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -15,10 +15,13 @@
//*****************************************************************************
#include "curl_downloader.hpp"

#include <array>
#include <cstdlib>
#include <ctime>
#include <filesystem>
#include <iostream>
#include <memory>
#include <mutex>

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Why do we need to include mutex in this PR?

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

std::once_flag and std::call_once

#include <string>

#include <curl/curl.h>
Expand All @@ -30,7 +33,7 @@

namespace ovms {

static const char* sizeUnits[] = {"B", "KB", "MB", "GB", "TB", NULL};
static constexpr std::array<const char*, 5> sizeUnits = {"B", "KB", "MB", "GB", "TB"};

static void print_download_speed_info(size_t received_size, size_t elapsed_time) {
double recv_len = (double)received_size;
Expand All @@ -39,20 +42,47 @@ static void print_download_speed_info(size_t received_size, size_t elapsed_time)
rate = elapsed ? recv_len / elapsed : received_size;

size_t rate_unit_idx = 0;
while (rate > 1000 && sizeUnits[rate_unit_idx + 1]) {
while (rate > 1000 && rate_unit_idx + 1 < sizeUnits.size()) {
rate /= 1000.0;
rate_unit_idx++;
}
printf(" [%.2f %s/s] ", rate, sizeUnits[rate_unit_idx]);
printf(" [%.2f %s/s] ", rate, sizeUnits.at(rate_unit_idx));
}

int computeProgressBarCells(size_t count, size_t max, int barWidth) {
if (max == 0 || barWidth <= 0) {
return 0;
}
const double ratio = static_cast<double>(count) / static_cast<double>(max);
if (ratio <= 0.0) {
return 0;
} else if (ratio >= 1.0) {
return barWidth;
}
return static_cast<int>(ratio * barWidth);
}

static void print_progress(size_t count, size_t max, bool first_run, size_t elapsed_time) {
// A response with no Content-Length reports dltotal == 0, so there is no ratio to show
if (max == 0) {
double received = (double)count;
size_t receivedUnitId = 0;
while (received > 1000 && receivedUnitId + 1 < sizeUnits.size()) {
received /= 1000.0;
receivedUnitId++;
Comment thread
mzegla marked this conversation as resolved.
}
printf("\rProgress: %.2f %s downloaded, total size unknown", received, sizeUnits.at(receivedUnitId));
print_download_speed_info(count, elapsed_time);
fflush(stdout);
return;
}

float progress = (float)count / max;
if (!first_run && progress < 0.01 && count > 0)
return;

const int bar_width = 50;
int bar_length = progress * bar_width;
const int bar_length = computeProgressBarCells(count, max, bar_width);

printf("\rProgress: [");
int i;
Expand All @@ -64,11 +94,11 @@ static void print_progress(size_t count, size_t max, bool first_run, size_t elap
}
size_t totalSizeUnitId = 0;
double totalSize = max;
while (totalSize > 1000 && sizeUnits[totalSizeUnitId + 1]) {
while (totalSize > 1000 && totalSizeUnitId + 1 < sizeUnits.size()) {
totalSize /= 1000.0;
totalSizeUnitId++;
}
printf("] %.2f%% of %.2f %s", progress * 100, totalSize, sizeUnits[totalSizeUnitId]);
printf("] %.2f%% of %.2f %s", progress * 100, totalSize, sizeUnits.at(totalSizeUnitId));
print_download_speed_info(count, elapsed_time);
if (progress == 1.0)
printf("\n");
Expand Down Expand Up @@ -117,6 +147,25 @@ static size_t file_write_callback(void* buffer, size_t size, size_t nmemb, void*
} \
} while (0)

// Keep one balanced libcurl global initialization for this downloader's process lifetime.
// The previous per-call guard held a null unique_ptr, so its deleter never ran and every
// download added another unmatched curl_global_init() call.
static Status ensureCurlGlobalInit() {
static std::once_flag initFlag;
static CURLcode initResult = CURLE_OK;
std::call_once(initFlag, []() {
initResult = curl_global_init(CURL_GLOBAL_DEFAULT);
if (initResult == CURLE_OK) {
std::atexit([]() { curl_global_cleanup(); });
}
Comment thread
rasapala marked this conversation as resolved.
});
if (initResult != CURLE_OK) {
SPDLOG_ERROR("curl error: {}. Error code: {}", curl_easy_strerror(initResult), (int)initResult);
return StatusCode::INTERNAL_ERROR;
}
return StatusCode::OK;
}
Comment on lines +150 to +167

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This means that every unit test that did start OVMS before did curl init and curl cleanup. So ovms behaves now more differently in ovms gtest scenarios than in prod. Why could we not just move this initialization to pull module?

Why we can't follow pattern used in httpservermodule.cpp?
Init curl in module init, cleanup in shutdown?

Then we could dispose whole part of curl_global_init from functions used here/.

Copy link
Copy Markdown
Collaborator Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

First question - yes.
We could follow the HttpServerModule init/shutdown pattern for the cases where HfPullModelModule::start()/shutdown() actually runs — but curl_downloader's functions aren't only invoked through that module's lifecycle. The curl_downloader.cpp-local std::call_once + atexit guard exists precisely to make the downloader self-sufficient regardless of how it's invoked. That said, it is true this changes gtest behavior (every test that exercises this file now does one process-wide curl_global_init/atexit-cleanup pair instead of zero) — but that's a correctness fix, not a new inconsistency: previously the guard was broken.

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

It is used only in pull_module. The other place we use it is:
https://github.com/openvinotoolkit/model_server/blob/main/src/llm/io_processing/image_utils.cpp#L68

but its entirely different curl call with their own guards.

Why did previous solution not work?
If there was issue with guards it could be reused solution from image_utils if you want to keep it self-contained. If we want to keep it called only once (which is not required by curl) then we could have it pull module.


struct ProgressData {
time_t started_download;
time_t last_print_time;
Expand Down Expand Up @@ -159,9 +208,10 @@ Status downloadFileWithCurl(const std::string& url, const std::string& filePath,
std::string agentString = std::string(PROJECT_NAME) + "/" + std::string(PROJECT_VERSION);

CURL* curl = nullptr;
CHECK_CURL_CALL(curl_global_init(CURL_GLOBAL_DEFAULT));
auto globalCurlGuard = std::unique_ptr<void, void (*)(void*)>(
nullptr, [](void*) { curl_global_cleanup(); });
auto initStatus = ensureCurlGlobalInit();
if (!initStatus.ok()) {
return initStatus;
}
curl = curl_easy_init();
if (!curl) {
SPDLOG_ERROR("Failed to initialize cURL.");
Expand Down Expand Up @@ -211,9 +261,10 @@ Status fetchUrlToString(const std::string& url, const std::string& authToken, st
std::string agentString = std::string(PROJECT_NAME) + "/" + std::string(PROJECT_VERSION);

CURL* curl = nullptr;
CHECK_CURL_CALL(curl_global_init(CURL_GLOBAL_DEFAULT));
auto globalCurlGuard = std::unique_ptr<void, void (*)(void*)>(
nullptr, [](void*) { curl_global_cleanup(); });
auto initStatus = ensureCurlGlobalInit();
if (!initStatus.ok()) {
return initStatus;
}
curl = curl_easy_init();
if (!curl) {
SPDLOG_ERROR("Failed to initialize cURL.");
Expand Down
5 changes: 5 additions & 0 deletions src/pull_module/curl_downloader.hpp
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//*****************************************************************************
#include <cstddef>
#include <string>

namespace ovms {
Expand All @@ -23,4 +24,8 @@ Status downloadFileWithCurl(const std::string& url, const std::string& filePath)
Status downloadFileWithCurl(const std::string& url, const std::string& filePath, const std::string& authTokenHF);
Status fetchUrlToString(const std::string& url, const std::string& authToken, std::string& responseBody);

// Number of filled cells in a barWidth-wide progress bar for count out of max bytes,
// clamped to [0, barWidth]. max == 0 means the server sent no Content-Length
int computeProgressBarCells(size_t count, size_t max, int barWidth);

} // namespace ovms
83 changes: 78 additions & 5 deletions src/test/pull_hf_model_test.cpp
Original file line number Diff line number Diff line change
Expand Up @@ -13,11 +13,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.
//*****************************************************************************
#include <algorithm>
#include <array>
#include <chrono>
#include <cstdlib>
#include <fstream>
#include <iomanip>
#include <limits>
#include <memory>
#include <openssl/sha.h>
#include <mutex>
Expand Down Expand Up @@ -47,6 +49,7 @@
#include "src/test/test_file_utils.hpp"
#include "src/test/test_with_temp_dir.hpp"
#include "src/filesystem/filesystem.hpp"
#include "src/pull_module/curl_downloader.hpp"
#include "src/pull_module/hf_pull_model_module.hpp"
#include "src/pull_module/libgit2.hpp"
#include "src/pull_module/optimum_export.hpp"
Expand Down Expand Up @@ -378,6 +381,76 @@ ::testing::AssertionResult interruptPosixWorkerAndExpectGracefulExit(pid_t child

} // namespace

TEST(CurlDownloaderProgressTest, UnknownTotalYieldsNoFilledCells) {
EXPECT_EQ(ovms::computeProgressBarCells(0, 0, 50), 0);
EXPECT_EQ(ovms::computeProgressBarCells(1024, 0, 50), 0);
EXPECT_EQ(ovms::computeProgressBarCells(std::numeric_limits<size_t>::max(), 0, 50), 0);
}

TEST(CurlDownloaderProgressTest, FilledCellsTrackRatio) {
EXPECT_EQ(ovms::computeProgressBarCells(0, 100, 50), 0);
EXPECT_EQ(ovms::computeProgressBarCells(50, 100, 50), 25);
EXPECT_EQ(ovms::computeProgressBarCells(100, 100, 50), 50);
}

TEST(CurlDownloaderProgressTest, FilledCellsClampToBarWidth) {
EXPECT_EQ(ovms::computeProgressBarCells(200, 100, 50), 50);
EXPECT_EQ(ovms::computeProgressBarCells(100, 100, 0), 0);
EXPECT_EQ(ovms::computeProgressBarCells(100, 100, -1), 0);
}

TEST_F(TestWithTempDir, ChunkedTransferWithoutContentLengthDownloadsFile) {
const std::string body(64 * 1024, 'x');
httplib::Server server;
server.Get("/chunked", [&body](const httplib::Request&, httplib::Response& res) {
res.set_chunked_content_provider("application/octet-stream",
[&body](size_t offset, httplib::DataSink& sink) {
if (offset >= body.size()) {
sink.done();
return true;
}
const size_t chunkSize = std::min<size_t>(4096, body.size() - offset);
// Keep the transfer active past the one-second progress throttle so the
// unknown-total path reaches print_progress() before the download completes.
std::this_thread::sleep_for(std::chrono::milliseconds(100));
sink.write(body.data() + offset, chunkSize);
return true;
});
});
const int port = server.bind_to_any_port("127.0.0.1");
ASSERT_GT(port, 0);
std::thread serverThread([&server]() {
server.listen_after_bind();
});
server.wait_until_ready();

const std::string url = "http://127.0.0.1:" + std::to_string(port) + "/chunked";
const std::string outputPath = directoryPath + "/downloaded.bin";

EnvGuard envGuard;
envGuard.unset("http_proxy");
envGuard.unset("https_proxy");
envGuard.unset("HTTP_PROXY");
envGuard.unset("HTTPS_PROXY");
envGuard.unset("no_proxy");
envGuard.unset("NO_PROXY");

testing::internal::CaptureStdout();
const ovms::Status downloadStatus = ovms::downloadFileWithCurl(url, outputPath);
const std::string output = testing::internal::GetCapturedStdout();

server.stop();
serverThread.join();

ASSERT_EQ(downloadStatus, ovms::StatusCode::OK);
EXPECT_THAT(output, ::testing::HasSubstr("total size unknown"));

std::ifstream downloadedFile(outputPath, std::ios::binary);
std::ostringstream downloadedContent;
downloadedContent << downloadedFile.rdbuf();
EXPECT_EQ(downloadedContent.str(), body);
}

// RAII helper class for managing log file lifecycle.
// Creates a log file path and automatically removes it on destruction.
class LogFileGuard {
Expand Down Expand Up @@ -2453,7 +2526,7 @@ TEST_F(HfPullModelModuleLoraTest, ResolveHfLoraFilenames) {
ovms::ImageGenerationGraphSettingsImpl graphSettings;
ovms::LoraAdapterSettings adapter;
adapter.alias = "pokemon";
adapter.sourceLora = "juliensimon/sd-pokemon-lora";
adapter.sourceLora = "MohamedAhmedAE/stable-diffusion-v1-5_lora_finetuning";
adapter.sourceType = ovms::LoraSourceType::HF_REPO;
graphSettings.loraAdapters.push_back(adapter);
settings.graphSettings = graphSettings;
Expand All @@ -2480,7 +2553,7 @@ TEST_F(HfPullModelModuleLoraTest, PullLoraAdaptersFromHfRepo) {
ovms::ImageGenerationGraphSettingsImpl graphSettings;
ovms::LoraAdapterSettings adapter;
adapter.alias = "pokemon";
adapter.sourceLora = "juliensimon/sd-pokemon-lora";
adapter.sourceLora = "MohamedAhmedAE/stable-diffusion-v1-5_lora_finetuning";
adapter.safetensorsFile = "pytorch_lora_weights.safetensors"; // explicit filename — skips HF API resolve
adapter.sourceType = ovms::LoraSourceType::HF_REPO;
graphSettings.loraAdapters.push_back(adapter);
Expand All @@ -2489,7 +2562,7 @@ TEST_F(HfPullModelModuleLoraTest, PullLoraAdaptersFromHfRepo) {
auto status = module.testPullLoraAdapters(this->directoryPath);
ASSERT_TRUE(status.ok()) << status.string();

auto loraFilePath = ovms::FileSystem::joinPath({this->directoryPath, "loras", "juliensimon/sd-pokemon-lora", "pytorch_lora_weights.safetensors"});
auto loraFilePath = ovms::FileSystem::joinPath({this->directoryPath, "loras", "MohamedAhmedAE/stable-diffusion-v1-5_lora_finetuning", "pytorch_lora_weights.safetensors"});
ASSERT_TRUE(std::filesystem::exists(loraFilePath)) << loraFilePath;
EXPECT_GT(std::filesystem::file_size(loraFilePath), 0);
}
Expand Down Expand Up @@ -2539,7 +2612,7 @@ TEST_F(HfDownloaderPullHfModel, DownloadImageGenModelWithLoRA) {
std::string modelName = "OpenVINO/stable-diffusion-v1-5-int8-ov";
std::string downloadPath = ovms::FileSystem::joinPath({this->directoryPath, "repository"});
std::string task = "image_generation";
std::string sourceLoras = "pokemon=juliensimon/sd-pokemon-lora@pytorch_lora_weights.safetensors";
std::string sourceLoras = "pokemon=MohamedAhmedAE/stable-diffusion-v1-5_lora_finetuning@pytorch_lora_weights.safetensors";
::SetUpServerForDownloadWithLoras(this->t, this->server, modelName, downloadPath, task, sourceLoras);

std::string basePath = ovms::FileSystem::joinPath({downloadPath, "OpenVINO", "stable-diffusion-v1-5-int8-ov"});
Expand All @@ -2550,7 +2623,7 @@ TEST_F(HfDownloaderPullHfModel, DownloadImageGenModelWithLoRA) {
ASSERT_TRUE(std::filesystem::exists(graphPath)) << graphPath;

// Verify LoRA adapter was downloaded
std::string loraDir = ovms::FileSystem::joinPath({basePath, "loras", "juliensimon", "sd-pokemon-lora"});
std::string loraDir = ovms::FileSystem::joinPath({basePath, "loras", "MohamedAhmedAE", "stable-diffusion-v1-5_lora_finetuning"});
auto loraFiles = searchFilesRecursively(loraDir, {"pytorch_lora_weights.safetensors"});
ASSERT_FALSE(loraFiles.empty()) << "LoRA .safetensors not found in: " << loraDir;

Expand Down
5 changes: 3 additions & 2 deletions third_party/libgit2/lfs.patch
Original file line number Diff line number Diff line change
Expand Up @@ -427,7 +427,7 @@ new file mode 100644
index 000000000..18490e5ad
--- /dev/null
+++ b/src/libgit2/lfs_filter.c
@@ -0,0 +1,2014 @@
@@ -0,0 +1,2015 @@
+/*
+/ Copyright 2025 Intel Corporation
+/
Expand Down Expand Up @@ -1565,7 +1565,8 @@ index 000000000..18490e5ad
+ return;
+
+ bar_width = 50;
+ bar_length = progress * bar_width;
+ /* Clamp: a server reporting more bytes than dltotal must not overflow bar_width. */
+ bar_length = (progress >= 1.0) ? bar_width : (int)(progress * bar_width);
+
+ printf("\rProgress: [");
+ for (i = 0; i < bar_length; ++i) {
Expand Down