Skip to content

Commit acba77b

Browse files
author
Sergey N. Yatskevich
committed
Add support of epoll and kqueue into Socket::select and SocketImpl::poll
1 parent ee893eb commit acba77b

3 files changed

Lines changed: 471 additions & 5 deletions

File tree

‎Foundation/include/Poco/Platform.h‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -184,6 +184,15 @@
184184
#define POCO_ARCH_BIG_ENDIAN 1
185185
#endif
186186

187+
//TODO: need to determine Linux > 2.6.0
188+
#if (POCO_OS == POCO_OS_LINUX)
189+
#define POCO_HAVE_FD_EPOLL 1
190+
#endif
191+
192+
//TODO: need to determine which of FreeBSD have kqueue
193+
#if (POCO_OS == POCO_OS_FREE_BSD)
194+
#define POCO_HAVE_FD_KQUEUE 1
195+
#endif
187196

188197
//TODO: determine all platforms having poll() call
189198
#if (POCO_OS == POCO_OS_SOLARIS) || (POCO_OS == POCO_OS_LINUX)

‎Net/src/Socket.cpp‎

Lines changed: 308 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -39,14 +39,23 @@
3939
#include "Poco/Timestamp.h"
4040
#include <algorithm>
4141
#include <string.h> // FD_SET needs memset on some platforms, so we can't use <cstring>
42-
#if defined(POCO_HAVE_FD_POLL)
42+
43+
44+
#if defined(POCO_HAVE_FD_EPOLL)
45+
#include <sys/epoll.h>
46+
#elif defined(POCO_HAVE_FD_KQUEUE)
47+
#include <sys/types.h>
48+
#include <sys/event.h>
49+
#include <sys/time.h>
50+
#elif defined(POCO_HAVE_FD_POLL)
4351
#include "Poco/SharedPtr.h"
4452
#include <poll.h>
4553
typedef Poco::SharedPtr<pollfd,
4654
Poco::ReferenceCounter,
4755
Poco::ReleaseArrayPolicy<pollfd> > SharedPollArray;
4856
#endif
4957

58+
5059
namespace Poco {
5160
namespace Net {
5261

@@ -93,7 +102,303 @@ Socket::~Socket()
93102

94103
int Socket::select(SocketList& readList, SocketList& writeList, SocketList& exceptList, const Poco::Timespan& timeout)
95104
{
96-
#if defined(POCO_HAVE_FD_POLL)
105+
#if defined(POCO_HAVE_FD_EPOLL)
106+
#warning "Poco use EPOLL for Socket::select"
107+
//
108+
// Size of epoll queue
109+
//
110+
int epoll_size = readList.size() + writeList.size() + exceptList.size();
111+
112+
//
113+
// If nothing to do, return 0
114+
//
115+
if (epoll_size == 0)
116+
return 0;
117+
118+
//
119+
// Fill epoll queue
120+
//
121+
int epollfd = -1;
122+
{
123+
//
124+
// Epoll events to be filled
125+
//
126+
struct epoll_event events_in[epoll_size];
127+
memset(events_in, 0, sizeof (events_in));
128+
129+
//
130+
// Current epoll event to be filled
131+
//
132+
struct epoll_event* event_last = events_in;
133+
134+
for (SocketList::iterator it = readList.begin(); it != readList.end(); ++it)
135+
{
136+
if (it->sockfd() != POCO_INVALID_SOCKET)
137+
{
138+
//
139+
// Try to find file descriptor in epoll events
140+
//
141+
struct epoll_event* e = events_in;
142+
for (; e != event_last; ++e)
143+
{
144+
if (reinterpret_cast<Socket*> (e->data.ptr)->sockfd() == it->sockfd())
145+
break;
146+
}
147+
148+
//
149+
// If not found allocate new epoll event
150+
//
151+
if (e == event_last)
152+
{
153+
e->data.ptr = &(*it);
154+
155+
++event_last;
156+
}
157+
158+
e->events |= EPOLLIN;
159+
}
160+
}
161+
162+
for (SocketList::iterator it = writeList.begin(); it != writeList.end(); ++it)
163+
{
164+
if (it->sockfd() != POCO_INVALID_SOCKET)
165+
{
166+
//
167+
// Try to find file descriptor in epoll events
168+
//
169+
struct epoll_event* e = events_in;
170+
for (; e != event_last; ++e)
171+
{
172+
if (reinterpret_cast<Socket*> (e->data.ptr)->sockfd() == it->sockfd())
173+
break;
174+
}
175+
176+
//
177+
// If not found allocate new epoll event
178+
//
179+
if (e == event_last)
180+
{
181+
e->data.ptr = &(*it);
182+
183+
++event_last;
184+
}
185+
186+
e->events |= EPOLLOUT;
187+
}
188+
}
189+
190+
for (SocketList::iterator it = exceptList.begin(); it != exceptList.end(); ++it)
191+
{
192+
if (it->sockfd() != POCO_INVALID_SOCKET)
193+
{
194+
//
195+
// Try to find file descriptor in epoll events
196+
//
197+
struct epoll_event* e = events_in;
198+
for (; e != event_last; ++e)
199+
{
200+
if (reinterpret_cast<Socket*> (e->data.ptr)->sockfd() == it->sockfd())
201+
break;
202+
}
203+
204+
//
205+
// If not found allocate new epoll event
206+
//
207+
if (e == event_last)
208+
{
209+
e->data.ptr = &(*it);
210+
211+
++event_last;
212+
}
213+
214+
e->events |= EPOLLERR;
215+
}
216+
}
217+
218+
//
219+
// Recalculate real epoll queue size
220+
//
221+
epoll_size = event_last - events_in;
222+
223+
//
224+
// Allocate epoll queue
225+
//
226+
epollfd = epoll_create(epoll_size);
227+
if (epollfd < 0)
228+
{
229+
char buf[4000];
230+
strerror_r(errno, buf, sizeof(buf));
231+
232+
SocketImpl::error(std::string("Can't create epoll - ") + buf);
233+
}
234+
235+
//
236+
// Place epoll events into epoll queue
237+
//
238+
for (struct epoll_event* e = events_in; e != event_last; ++e)
239+
{
240+
if (epoll_ctl(epollfd, EPOLL_CTL_ADD, reinterpret_cast<Socket*> (e->data.ptr)->sockfd(), e) < 0)
241+
{
242+
char buf[4000];
243+
strerror_r(errno, buf, sizeof(buf));
244+
245+
::close(epollfd);
246+
SocketImpl::error(std::string("Can't insert socket to epoll - ") + buf);
247+
}
248+
}
249+
}
250+
251+
struct epoll_event events_out[epoll_size];
252+
memset(events_out, 0, sizeof (events_out));
253+
254+
Poco::Timespan remainingTime(timeout);
255+
int rc;
256+
do
257+
{
258+
Poco::Timestamp start;
259+
rc = epoll_wait(epollfd, events_out, epoll_size, remainingTime.totalMilliseconds());
260+
if (rc < 0 && SocketImpl::lastError() == POCO_EINTR)
261+
{
262+
Poco::Timestamp end;
263+
Poco::Timespan waited = end - start;
264+
if (waited < remainingTime)
265+
remainingTime -= waited;
266+
else
267+
remainingTime = 0;
268+
}
269+
}
270+
while (rc < 0 && SocketImpl::lastError() == POCO_EINTR);
271+
272+
//
273+
// Close epoll queue
274+
//
275+
::close(epollfd);
276+
277+
if (rc < 0) SocketImpl::error();
278+
279+
SocketList readyReadList;
280+
SocketList readyWriteList;
281+
SocketList readyExceptList;
282+
for (int n = 0; n < rc; ++n)
283+
{
284+
if (events_out[n].events & EPOLLERR)
285+
readyExceptList.push_back(*reinterpret_cast<Socket*>(events_out[n].data.ptr));
286+
if (events_out[n].events & EPOLLIN)
287+
readyReadList.push_back(*reinterpret_cast<Socket*>(events_out[n].data.ptr));
288+
if (events_out[n].events & EPOLLOUT)
289+
readyWriteList.push_back(*reinterpret_cast<Socket*>(events_out[n].data.ptr));
290+
}
291+
std::swap(readList, readyReadList);
292+
std::swap(writeList, readyWriteList);
293+
std::swap(exceptList, readyExceptList);
294+
return readList.size() + writeList.size() + exceptList.size();
295+
296+
#elif defined(POCO_HAVE_FD_KQUEUE)
297+
#warning "Poco use KQUEUE for Socket::select"
298+
//
299+
// Size of kqueue queue
300+
//
301+
int kqueue_size = readList.size() + writeList.size() + exceptList.size();
302+
303+
//
304+
// If nothing to do, return 0
305+
//
306+
if (kqueue_size == 0)
307+
return 0;
308+
309+
//
310+
// Create kevent queue
311+
//
312+
int kqueuefd = kqueue();
313+
if (kqueuefd < 0)
314+
{
315+
char buf[4000];
316+
strerror_r(errno, buf, sizeof(buf));
317+
318+
SocketImpl::error(std::string("Can't create kqueue - ") + buf);
319+
}
320+
321+
//
322+
// Allocate in/out kevent queues
323+
//
324+
struct kevent events_in[kqueue_size];
325+
struct kevent events_out[kqueue_size];
326+
memset(&events_in , 0, sizeof(events_in));
327+
memset(&events_out, 0, sizeof(events_out));
328+
329+
//
330+
// Add sockets to events_in list for appropriate event
331+
//
332+
for (size_t i = 0; i < readList.size(); ++i)
333+
{
334+
if (readList[i].sockfd () != POCO_INVALID_SOCKET)
335+
{
336+
EV_SET(events_in + i, readList[i].sockfd (), EVFILT_READ, EV_ADD|EV_CLEAR, 0, 0, &readList[i]);
337+
}
338+
}
339+
for (size_t i = 0; i < writeList.size(); ++i)
340+
{
341+
if (writeList[i].sockfd () != POCO_INVALID_SOCKET)
342+
{
343+
EV_SET(events_in + readList.size () + i, writeList[i].sockfd (), EVFILT_WRITE, EV_ADD|EV_CLEAR, 0, 0, &writeList[i]);
344+
}
345+
}
346+
for (size_t i = 0; i < exceptList.size(); ++i)
347+
{
348+
if (exceptList[i].sockfd () != POCO_INVALID_SOCKET)
349+
{
350+
EV_SET(events_in + readList.size () + writeList.size () + i, exceptList[i].sockfd (), EVFILT_READ/*FIXME*/, EV_ADD|EV_CLEAR, 0, 0, &exceptList[i]);
351+
}
352+
}
353+
354+
Poco::Timespan remainingTime(timeout);
355+
int rc;
356+
do
357+
{
358+
struct timespec ts;
359+
ts.tv_sec = (long)remainingTime.totalSeconds();
360+
ts.tv_nsec = (long)remainingTime.useconds();
361+
362+
Poco::Timestamp start;
363+
rc = kevent(kqueuefd, events_in, kqueue_size, events_out, kqueue_size, &ts);
364+
if (rc < 0 && SocketImpl::lastError() == POCO_EINTR)
365+
{
366+
Poco::Timestamp end;
367+
Poco::Timespan waited = end - start;
368+
if (waited < remainingTime)
369+
remainingTime -= waited;
370+
else
371+
remainingTime = 0;
372+
}
373+
}
374+
while (rc < 0 && SocketImpl::lastError() == POCO_EINTR);
375+
376+
//
377+
// Close kqueue
378+
//
379+
::close(kqueuefd);
380+
381+
if (rc < 0) SocketImpl::error();
382+
383+
SocketList readyReadList;
384+
SocketList readyWriteList;
385+
SocketList readyExceptList;
386+
for (int n = 0; n < rc; ++n)
387+
{
388+
if (events_out[n].flags & EV_ERROR)
389+
readyExceptList.push_back(*reinterpret_cast<Socket*>(events_out[n].udata));
390+
else if (events_out[n].filter == EVFILT_READ)
391+
readyReadList.push_back(*reinterpret_cast<Socket*>(events_out[n].udata));
392+
else if (events_out[n].filter == EVFILT_WRITE)
393+
readyWriteList.push_back(*reinterpret_cast<Socket*>(events_out[n].udata));
394+
}
395+
std::swap(readList, readyReadList);
396+
std::swap(writeList, readyWriteList);
397+
std::swap(exceptList, readyExceptList);
398+
return readList.size() + writeList.size() + exceptList.size();
399+
400+
#elif defined(POCO_HAVE_FD_POLL)
401+
#warning "Poco use POLL for Socket::select"
97402
nfds_t nfd = readList.size() + writeList.size() + exceptList.size();
98403
if (0 == nfd) return 0;
99404

@@ -177,6 +482,7 @@ int Socket::select(SocketList& readList, SocketList& writeList, SocketList& exce
177482
return readList.size() + writeList.size() + exceptList.size();
178483

179484
#else
485+
#warning "Poco use SELECT for Socket::select"
180486
fd_set fdRead;
181487
fd_set fdWrite;
182488
fd_set fdExcept;

0 commit comments

Comments
 (0)