Here is how it happens:
Use curl_multi integration using SOCKETFUNCTION and TIMERFUNCTION.
Create many easy handle requesting through HTTP onto an HTTP server at
once.
Have the backlog configuration through "listen" on the server be lower
than the number requests.
Curl will create as many sockets as there are easy handles. And it will
request CURL_POLL_OUT for all of them. That is the way it know when it
is connected.
Only a fraction will get notified this CURL_CSELECT_OUT on the first
polls. The requests will go through.
At some point other sockets will start to connect and we notify curl.
But curl just asks to remove the socket from polling, and it never
sends the request.
I have made an example for Linux (epoll) that fails to complete on cURL
7.55.0 and 7.55.1, but succeeds in 7.54.1, though sometimes it can get
a bit slow.
--
Valentin David
This email reflects my own opinion. You have the right to disagree.
import socket
with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
s.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
s.bind(('', 5432))
s.listen(10)
i = 0
while True:
i = i + 1
conn, addr = s.accept()
data = b''
with conn:
while b'\r\n\r\n' not in data:
new_data = conn.recv(1024)
if len(new_data) == 0:
break
data = data + new_data
if b'\r\n\r\n' not in data:
print("NO REQUEST", data)
continue
tosend = '{}\r\n'.format(i).encode('utf-8')
conn.sendall(b'HTTP/1.1 200 OK\r\n')
conn.sendall('Content-Length: {}\r\n\r\n'.format(len(tosend)).encode('utf-8'))
conn.sendall(tosend)
conn.close()
#include <unistd.h>
#include <sys/epoll.h>
#include <curl/curl.h>
#include <vector>
#include <map>
#include <iostream>
#include <set>
struct pollset {
public:
pollset() {
epollfd = epoll_create1(0);
timeout_ms = 0;
multi = curl_multi_init();
curl_multi_setopt(multi, CURLMOPT_SOCKETFUNCTION, &pollset::socket_callback);
curl_multi_setopt(multi, CURLMOPT_SOCKETDATA, this);
curl_multi_setopt(multi, CURLMOPT_TIMERFUNCTION, &pollset::timer_callback);
curl_multi_setopt(multi, CURLMOPT_TIMERDATA, this);
}
~pollset() {
curl_multi_cleanup(multi);
close(epollfd);
}
void add(CURL* handle) {
curl_multi_add_handle(multi, handle);
}
void remove(CURL* handle) {
curl_multi_remove_handle(multi, handle);
}
void set_event(int socket, uint32_t events) {
auto inserted = handled_fds.insert(socket);
auto op = (inserted.second)?EPOLL_CTL_ADD:EPOLL_CTL_MOD;
epoll_event e;
e.events = events;
e.data.fd = socket;
epoll_ctl(epollfd, op, socket, &e);
}
void remove_event(int socket) {
handled_fds.erase(socket);
epoll_ctl(epollfd, EPOLL_CTL_DEL, socket, nullptr);
}
void set_timeout(int timeout_ms) {
this->timeout_ms = timeout_ms;
}
CURLMsg *info_read(int& msgs_in_queue) {
return curl_multi_info_read(multi, &msgs_in_queue);
}
void poll() {
epoll_event evt;
auto status = epoll_wait(epollfd, &evt, 1, timeout_ms);
if (status == 0) {
perform_action(CURL_SOCKET_TIMEOUT, 0);
} else {
int action = 0;
if (evt.events & EPOLLIN)
action |= CURL_CSELECT_IN;
if (evt.events & EPOLLOUT)
action |= CURL_CSELECT_OUT;
if (evt.events & EPOLLERR)
action |= CURL_CSELECT_ERR;
perform_action(evt.data.fd, action);
}
}
private:
int perform_action(int fd, int action) {
int running;
while (true) {
auto ret = curl_multi_socket_action(multi, fd, action,
&running);
if (ret != CURLM_CALL_MULTI_PERFORM)
break ;
}
return running;
}
private:
static int socket_callback(CURL *easy,
curl_socket_t s,
int what,
void* userp,
void* /*socketp*/) {
auto polls = static_cast<pollset*>(userp);
switch (what) {
case CURL_POLL_IN:
polls->set_event(s, EPOLLIN);
break ;
case CURL_POLL_OUT:
polls->set_event(s, EPOLLOUT);
break ;
case CURL_POLL_INOUT:
polls->set_event(s, EPOLLIN | EPOLLOUT);
break ;
case CURL_POLL_REMOVE:
polls->remove_event(s);
default:
break ;
}
return 0;
}
static int timer_callback(CURLM *multi,
long timeout_ms,
void *userp) {
auto polls = static_cast<pollset*>(userp);
polls->set_timeout(timeout_ms);
}
CURLM* multi;
int timeout_ms;
std::set<int> handled_fds;
int epollfd;
};
int main(void)
{
pollset p;
std::vector<CURL*> handles;
for (unsigned i = 0; i < 100; i++) {
handles.push_back(curl_easy_init());
curl_easy_setopt(handles[i], CURLOPT_URL, "http://localhost:5432/";);
curl_easy_setopt(handles[i], CURLOPT_VERBOSE, 1L);
p.add(handles[i]);
}
auto requests = handles.size();
while (requests > 0) {
p.poll();
while (true) {
int msgq = 0;
auto m = p.info_read(msgq);
if (!m)
break ;
if(m->msg == CURLMSG_DONE) {
--requests;
}
}
}
for (auto& h : handles) {
p.remove(h);
curl_easy_cleanup(h);
}
return 0;
}
-------------------------------------------------------------------
Unsubscribe: https://cool.haxx.se/list/listinfo/curl-library
Etiquette: https://curl.haxx.se/mail/etiquette.html
