Uh oh!
There was an error while loading. Please reload this page.
- Notifications
You must be signed in to change notification settings - Fork 1
Expand file tree
/
Copy pathflex_io.cpp
More file actions
Latest commit
175 lines (157 loc) · 6.4 KB
/
Copy pathflex_io.cpp
File metadata and controls
175 lines (157 loc) · 6.4 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
#include"flex_io.h"
#include"qgenlib/qgen_error.h"
#include<mutex>
std::unique_ptr<FlexReader> FlexReaderFactory::create_reader(constchar* path) {
std::unique_ptr<FlexReader> reader;
std::string path_str(path);
if ( path_str.rfind("s3://", 0) == 0 ) {
reader = std::make_unique<FlexHttpReader>();
const std::string prefix = "s3://";
size_t slash_pos = path_str.find('/', prefix.length());
if (slash_pos == std::string::npos) returnnullptr; // Invalid S3 URL
std::string bucket = path_str.substr(prefix.length(), slash_pos - prefix.length());
std::string key = path_str.substr(slash_pos + 1);
path_str = "https://" + bucket + ".s3.amazonaws.com/" + key;
//notice("path_str = %s", path_str.c_str());
}
elseif (path_str.rfind("http://", 0) == 0 || path_str.rfind("https://", 0) == 0) {
reader = std::make_unique<FlexHttpReader>();
} else {
reader = std::make_unique<FlexFileReader>();
}
if (reader && reader->open(path_str.c_str())) {
return reader;
}
else {
error("Failed to open reader for path: %s", path_str.c_str());
returnnullptr;
}
}
FlexFileReader::FlexFileReader() {}
FlexFileReader::~FlexFileReader() { close(); }
boolFlexFileReader::open(constchar* uri) {
path_.assign(uri);
fp_ = std::fopen(path_.c_str(), "rb");
if ( !fp_ ) {
error("Failed to open file %s", uri);
returnfalse;
}
std::fseek(fp_, 0, SEEK_END);
size_ = std::ftell(fp_);
std::fseek(fp_, 0, SEEK_SET);
returntrue;
}
boolFlexFileReader::read_at(uint64_t offset, uint64_t length, std::string& buffer) {
if ( !is_open() ) returnfalse;
if (std::fseek(fp_, static_cast<long>(offset), SEEK_SET) != 0) returnfalse;
buffer.resize(length);
size_t got = std::fread((void*)buffer.c_str(), 1, length, fp_);
if ( got != length ) {
buffer.clear();
returnfalse;
}
returntrue;
}
std::unique_ptr<FlexReader> FlexFileReader::clone() const {
auto reader = std::make_unique<FlexFileReader>();
if (!reader->open(path_.c_str())) {
returnnullptr;
}
return reader;
}
boolFlexHttpReader::parse_head() {
CURL* c = curl_easy_init();
if (!c) returnfalse;
curl_easy_setopt(c, CURLOPT_URL, url_.c_str());
curl_easy_setopt(c, CURLOPT_NOBODY, 1L);
curl_easy_setopt(c, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(c, CURLOPT_MAXREDIRS, 5L);
auto rc = curl_easy_perform(c);
long code = 0; curl_easy_getinfo(c, CURLINFO_RESPONSE_CODE, &code);
double cl = 0; curl_easy_getinfo(c, CURLINFO_CONTENT_LENGTH_DOWNLOAD, &cl);
curl_easy_cleanup(c);
if (rc != CURLE_OK) returnfalse;
if (code == 200 || code == 204) { if (cl > 0) size_ = static_cast<uint64_t>(cl); returntrue; }
returnfalse;
}
size_tFlexHttpReader::write_to_string(void* p, size_t sz, size_t nm, void* ud) {
//notice("write_to_string called with sz=%zu, nm=%zu", sz, nm);
auto* s = static_cast<std::string*>(ud);
s->append(static_cast<char*>(p), sz * nm);
return sz * nm;
// size_t new_length = size * nmemb;
// try { s->append((char*)contents, new_length); }
// catch (std::bad_alloc& e) { return 0; }
// return new_length;
}
FlexHttpReader::FlexHttpReader() {
// curl_global_init is not thread-safe and must be called exactly once for the
// whole process. With multiple FlexHttpReader instances (e.g. one per worker
// thread), per-instance init/cleanup would race and free shared state while
// other handles are still alive. Initialize once; rely on process exit for
// global cleanup (standard for a short-lived CLI).
static std::once_flag curl_global_once;
std::call_once(curl_global_once, []() { curl_global_init(CURL_GLOBAL_DEFAULT); });
curl_ = curl_easy_init();
}
FlexHttpReader::~FlexHttpReader() {
close();
}
boolFlexHttpReader::open(constchar* uri) {
//curl_ = curl_easy_init();
// if (!curl_) return false;
// curl_easy_setopt(curl_, CURLOPT_FOLLOWLOCATION, 1L);
// curl_easy_setopt(curl_, CURLOPT_MAXREDIRS, 5L);
// curl_easy_setopt(curl_, CURLOPT_TCP_KEEPALIVE, 1L);
// parse_head(); // optional
// is_open_ = true;
// return true;
if (!curl_) returnfalse;
url_.assign(uri);
curl_easy_setopt(curl_, CURLOPT_URL, url_.c_str());
curl_easy_setopt(curl_, CURLOPT_FOLLOWLOCATION, 1L);
curl_easy_setopt(curl_, CURLOPT_UNRESTRICTED_AUTH, 1L);
curl_easy_setopt(curl_, CURLOPT_NOBODY, 1L); // HEAD request
if (curl_easy_perform(curl_) == CURLE_OK) {
curl_off_t length = -1;
curl_easy_getinfo(curl_, CURLINFO_CONTENT_LENGTH_DOWNLOAD_T, &length);
if (length > 0) {
size_ = static_cast<uint64_t>(length);
is_open_ = true;
}
}
curl_easy_setopt(curl_, CURLOPT_NOBODY, 0L); // Reset for future GETs
return is_open_;
}
boolFlexHttpReader::read_at(uint64_t offset, uint64_t length, std::string& buffer) {
if (!curl_) returnfalse;
//std::string range = "bytes=" + std::to_string(offset) + "-" + std::to_string(offset + length - 1);
std::string range = std::to_string(offset) + "-" + std::to_string(offset + length - 1);
//notice("Requesting url: %s", url_.c_str());
//notice("Requesting range: %s", range.c_str());
buffer.clear();
curl_easy_setopt(curl_, CURLOPT_URL, url_.c_str());
curl_easy_setopt(curl_, CURLOPT_RANGE, range.c_str());
curl_easy_setopt(curl_, CURLOPT_WRITEFUNCTION, write_to_string);
curl_easy_setopt(curl_, CURLOPT_WRITEDATA, &buffer);
int32_t max_attempt = 4;
for (int32_t attempt = 0; attempt < max_attempt; ++attempt) {
auto rc = curl_easy_perform(curl_);
long code = 0;
curl_easy_getinfo(curl_, CURLINFO_RESPONSE_CODE, &code);
//notice("attempt %d: code = %ld", attempt, code);
if (rc == CURLE_OK && (code == 206 || (code == 200 && length == buffer.size())) && buffer.size() == length) returntrue;
//notice("buffer.size() = %zu, expected length = %zu", buffer.size(), length);
if (code == 429 || code == 503) { std::this_thread::sleep_for(std::chrono::milliseconds(150 << attempt)); continue; }
break;
}
buffer.clear();
returnfalse;
}
std::unique_ptr<FlexReader> FlexHttpReader::clone() const {
auto reader = std::make_unique<FlexHttpReader>();
if (!reader->open(url_.c_str())) {
returnnullptr;
}
return reader;
}