coffee/wepoll
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

epoll: split out socket related code

Browse Source
This commit is contained in:
Bert Belder 2017-09-07 05:07:50 +02:00
parent 1a8c20ef36
commit 0d941fd52a
7 changed files with 431 additions and 397 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

299
src/epoll-socket.c Normal file
View File

@ -0,0 +1,299 @@
#include <assert.h>
#include <malloc.h>
#include <stdbool.h>
#include <stdint.h>
#include "afd.h"
#include "epoll-socket.h"
#include "epoll.h"
#include "error.h"
#include "poll-request.h"
#include "port-data.h"
#ifndef SIO_BASE_HANDLE
#define SIO_BASE_HANDLE 0x48000022
#endif
#define _EP_EVENT_MASK 0xffff
#define _EP_SOCK_LISTED 0x1
#define _EP_SOCK_DELETED 0x2
#define ATTN_LIST_ADD(p, s) \
do { \
QUEUE_INSERT_TAIL(&(p)->update_queue, &(s)->queue_entry); \
(s)->flags |= _EP_SOCK_LISTED; \
} while (0)
static inline ep_sock_t* _ep_sock_alloc(void) {
ep_sock_t* sock_info = malloc(sizeof *sock_info);
if (sock_info == NULL)
return_error(NULL, ERROR_NOT_ENOUGH_MEMORY);
return sock_info;
}
static inline void _ep_sock_free(ep_sock_t* sock_info) {
free(sock_info);
}
ep_sock_t* ep_sock_new(_ep_port_data_t* port_data) {
ep_sock_t* sock_info = _ep_sock_alloc();
if (sock_info == NULL)
return NULL;
(void) port_data;
memset(sock_info, 0, sizeof *sock_info);
QUEUE_INIT(&sock_info->queue_entry);
return sock_info;
}
int ep_sock_delete(_ep_port_data_t* port_data, ep_sock_t* sock_info) {
/* Remove from lookup tree. */
if (sock_info->socket != 0 && !(sock_info->flags & _EP_SOCK_DELETED)) {
RB_REMOVE(ep_sock_tree, &port_data->sock_data_tree, sock_info);
sock_info->flags |= _EP_SOCK_DELETED;
}
/* Remove from attention list. */
if (sock_info->flags & _EP_SOCK_LISTED) {
QUEUE_REMOVE(&sock_info->queue_entry);
sock_info->flags &= ~_EP_SOCK_LISTED;
}
/* The socket may still have pending overlapped requests that have yet to be
* reported by the completion port. If that's the case the structure can't be
* free'd yet; epoll_wait() will do it as it dequeues those requests.
*/
if (sock_info->poll_req_count == 0)
_ep_sock_free(sock_info);
return 0;
}
int ep_sock_compare(ep_sock_t* a, ep_sock_t* b) {
return a->socket - b->socket;
}
static inline bool _ep_sock_delete_pending(ep_sock_t* sock_info) {
return sock_info->flags & _EP_SOCK_DELETED;
}
void ep_sock_register_poll_req(_ep_port_data_t* port_data,
ep_sock_t* sock_info) {
assert(!_ep_sock_delete_pending(sock_info));
sock_info->poll_req_count++;
port_data->poll_req_count++;
}
void ep_sock_unregister_poll_req(_ep_port_data_t* port_data,
ep_sock_t* sock_info) {
sock_info->poll_req_count--;
port_data->poll_req_count--;
if (_ep_sock_delete_pending(sock_info) && sock_info->poll_req_count == 0)
ep_sock_delete(port_data, sock_info);
}
static int _get_related_sockets(_ep_port_data_t* port_data,
SOCKET socket,
SOCKET* afd_socket_out,
SOCKET* driver_socket_out) {
SOCKET afd_socket, driver_socket;
DWORD bytes;
/* Try to obtain a base handle for the socket, so we can bypass LSPs
* that get in the way if we want to talk to the kernel directly. If
* it fails we try if we work with the original socket. Note that on
* windows XP/2k3 this will always fail since they don't support the
* SIO_BASE_HANDLE ioctl.
*/
afd_socket = socket;
WSAIoctl(socket,
SIO_BASE_HANDLE,
NULL,
0,
&afd_socket,
sizeof afd_socket,
&bytes,
NULL,
NULL);
driver_socket = _ep_get_driver_socket(port_data, afd_socket);
if (driver_socket == INVALID_SOCKET)
return -1;
*afd_socket_out = afd_socket;
*driver_socket_out = driver_socket;
return 0;
}
int ep_sock_set_socket(_ep_port_data_t* port_data,
ep_sock_t* sock_info,
SOCKET socket) {
if (socket == 0 || socket == INVALID_SOCKET)
return_error(-1, ERROR_INVALID_HANDLE);
if (sock_info->socket != 0)
return_error(-1, ERROR_INVALID_PARAMETER);
if (_get_related_sockets(port_data,
socket,
&sock_info->afd_socket,
&sock_info->driver_socket) < 0)
return -1;
sock_info->socket = socket;
if (RB_INSERT(ep_sock_tree, &port_data->sock_data_tree, sock_info) != NULL) {
sock_info->socket = 0;
return_error(-1, ERROR_ALREADY_EXISTS); /* Socket already in epoll set. */
}
return 0;
}
int ep_sock_set_event(_ep_port_data_t* port_data,
ep_sock_t* sock_info,
const struct epoll_event* ev) {
/* EPOLLERR and EPOLLHUP are always reported, even when no sollicited. */
uint32_t events = ev->events | EPOLLERR | EPOLLHUP;
sock_info->user_events = events;
sock_info->user_data = ev->data;
if (events & _EP_EVENT_MASK & ~(sock_info->latest_poll_req_events)) {
/* Add to attention list, if not already added. */
if (!(sock_info->flags & _EP_SOCK_LISTED)) {
QUEUE_INSERT_TAIL(&port_data->update_queue, &sock_info->queue_entry);
sock_info->flags |= _EP_SOCK_LISTED;
}
}
return 0;
}
static inline bool _is_latest_poll_req(ep_sock_t* sock_info,
poll_req_t* poll_req) {
return poll_req == sock_info->latest_poll_req;
}
static inline void _clear_latest_poll_req(ep_sock_t* sock_info) {
sock_info->latest_poll_req = NULL;
sock_info->latest_poll_req_events = 0;
}
static inline void _set_latest_poll_req(ep_sock_t* sock_info,
poll_req_t* poll_req,
uint32_t epoll_events) {
sock_info->latest_poll_req = poll_req;
sock_info->latest_poll_req_events = epoll_events;
}
static int _ep_submit_poll_req(_ep_port_data_t* port_data,
ep_sock_t* sock_info) {
poll_req_t* poll_req;
uint32_t epoll_events = sock_info->user_events;
poll_req = poll_req_new(port_data, sock_info);
if (poll_req == NULL)
return -1;
if (poll_req_submit(poll_req,
epoll_events,
sock_info->afd_socket,
sock_info->driver_socket) < 0) {
poll_req_delete(port_data, sock_info, poll_req);
return -1;
}
_set_latest_poll_req(sock_info, poll_req, epoll_events);
return 0;
}
int ep_sock_update(_ep_port_data_t* port_data, ep_sock_t* sock_info) {
bool broken = false;
assert(sock_info->flags & _EP_SOCK_LISTED);
/* Check if there are events registered that are not yet submitted. In
* that case we need to submit another req.
*/
if ((sock_info->user_events & _EP_EVENT_MASK &
~sock_info->latest_poll_req_events) == 0)
/* All the events the user is interested in are already being monitored
* by the latest poll request. */
goto done;
if (_ep_submit_poll_req(port_data, sock_info) < 0) {
if (GetLastError() == ERROR_INVALID_HANDLE)
/* The socket is broken. It will be dropped from the epoll set. */
broken = true;
else
/* Another error occurred, which is propagated to the caller. */
return -1;
}
done:
/* Remove the socket from the update queue. */
QUEUE_REMOVE(&sock_info->queue_entry);
sock_info->flags &= ~_EP_SOCK_LISTED;
/* If we saw an ERROR_INVALID_HANDLE error, drop the socket. */
if (broken)
ep_sock_delete(port_data, sock_info);
return 0;
}
int ep_sock_feed_event(_ep_port_data_t* port_data,
poll_req_t* poll_req,
struct epoll_event* ev) {
ep_sock_t* sock_info = poll_req_get_sock_data(poll_req);
uint32_t epoll_events;
bool drop_socket;
int ev_count = 0;
if (_ep_sock_delete_pending(sock_info) ||
!_is_latest_poll_req(sock_info, poll_req)) {
/* Ignore completion for overlapped poll operation if it isn't
* the the most recently posted one, or if the socket has been
* deleted. */
poll_req_delete(port_data, sock_info, poll_req);
return 0;
}
_clear_latest_poll_req(sock_info);
poll_req_complete(poll_req, &epoll_events, &drop_socket);
/* Filter events that the user didn't ask for. */
epoll_events &= sock_info->user_events;
/* Drop the socket if the EPOLLONESHOT flag is set and there are any events
* to report. */
if (epoll_events != 0 && (sock_info->user_events & EPOLLONESHOT))
drop_socket = true;
/* Fill the ev structure if there are any events to report. */
if (epoll_events != 0) {
ev->data = sock_info->user_data;
ev->events = epoll_events;
ev_count = 1;
}
poll_req_delete(port_data, sock_info, poll_req);
if (drop_socket)
/* Drop the socket from the epoll set. */
ep_sock_delete(port_data, sock_info);
else
/* Put the socket back onto the attention list so a new poll request will
* be submitted. */
ATTN_LIST_ADD(port_data, sock_info);
return ev_count;
}

56
src/epoll-socket.h Normal file
View File

@ -0,0 +1,56 @@
#ifndef EPOLL_SOCK_DATA_H_
#define EPOLL_SOCK_DATA_H_
#include <stdint.h>
#include "epoll.h"
#include "queue.h"
#include "tree.h"
#include "util.h"
#include "win.h"
typedef struct _ep_port_data _ep_port_data_t;
typedef struct poll_req poll_req_t;
typedef RB_HEAD(ep_sock_tree, ep_sock) ep_sock_tree_head_t;
typedef RB_ENTRY(ep_sock) ep_sock_tree_entry_t;
typedef struct ep_sock {
SOCKET socket;
SOCKET afd_socket;
SOCKET driver_socket;
ep_sock_tree_entry_t tree_entry;
QUEUE queue_entry;
epoll_data_t user_data;
poll_req_t* latest_poll_req;
uint32_t user_events;
uint32_t latest_poll_req_events;
uint32_t poll_req_count;
uint32_t flags;
} ep_sock_t;
ep_sock_t* ep_sock_new(_ep_port_data_t* port_data);
int ep_sock_delete(_ep_port_data_t* port_data, ep_sock_t* sock_info);
int ep_sock_set_socket(_ep_port_data_t* port_data,
ep_sock_t* sock_info,
SOCKET socket);
int ep_sock_set_event(_ep_port_data_t* port_data,
ep_sock_t* sock_info,
const struct epoll_event* ev);
int ep_sock_update(_ep_port_data_t* port_data, ep_sock_t* sock_info);
int ep_sock_feed_event(_ep_port_data_t* port_data,
poll_req_t* poll_req,
struct epoll_event* ev);
void ep_sock_register_poll_req(_ep_port_data_t* port_data,
ep_sock_t* sock_info);
void ep_sock_unregister_poll_req(_ep_port_data_t* port_data,
ep_sock_t* sock_info);
int ep_sock_compare(ep_sock_t* a, ep_sock_t* b);
RB_GENERATE_STATIC(ep_sock_tree, ep_sock, tree_entry, ep_sock_compare)
#endif /* EPOLL_SOCK_DATA_H_ */

410
src/epoll.c
View File

@ -6,10 +6,12 @@
#include <stdlib.h>
#include "afd.h"
#include "epoll-socket.h"
#include "epoll.h"
#include "error.h"
#include "nt.h"
#include "poll-request.h"
#include "port-data.h"
#include "queue.h"
#include "tree.h"
#include "util.h"
@ -17,235 +19,28 @@
#define _EP_COMPLETION_LIST_LENGTH 64
#ifndef SIO_BASE_HANDLE
#define SIO_BASE_HANDLE 0x48000022
#endif
#define _EP_EVENT_MASK 0xffff
#define _EP_SOCK_LISTED 0x1
#define _EP_SOCK_DELETED 0x2
#define ATTN_LIST_ADD(p, s) \
do { \
QUEUE_INSERT_TAIL(&(p)->update_queue, &(s)->queue_entry); \
(s)->flags |= _EP_SOCK_LISTED; \
} while (0)
typedef struct _ep_port_data _ep_port_data_t;
typedef struct poll_req poll_req_t;
typedef struct _ep_sock_data _ep_sock_data_t;
typedef struct ep_sock ep_sock_t;
static int _ep_initialize(void);
static SOCKET _ep_get_driver_socket(_ep_port_data_t* port_data,
WSAPROTOCOL_INFOW* protocol_info);
static SOCKET _ep_create_driver_socket(HANDLE iocp,
WSAPROTOCOL_INFOW* protocol_info);
static int _ep_compare_sock_data(_ep_sock_data_t* a, _ep_sock_data_t* b);
static int _ep_submit_poll_req(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data);
ep_sock_t* sock_info);
static int _ep_initialized = 0;
/* State associated with a epoll handle. */
typedef struct _ep_port_data {
HANDLE iocp;
SOCKET driver_sockets[array_count(AFD_PROVIDER_GUID_LIST)];
RB_HEAD(_ep_sock_data_tree, _ep_sock_data) sock_data_tree;
QUEUE update_queue;
size_t poll_req_count;
} _ep_port_data_t;
/* State associated with a socket that is registered to the epoll port. */
typedef struct _ep_sock_data {
SOCKET socket;
SOCKET afd_socket;
SOCKET driver_socket;
RB_ENTRY(_ep_sock_data) tree_entry;
QUEUE queue_entry;
epoll_data_t user_data;
poll_req_t* latest_poll_req;
uint32_t user_events;
uint32_t latest_poll_req_events;
uint32_t poll_req_count;
uint32_t flags;
} _ep_sock_data_t;
static int _ep_sock_data_delete(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data);
RB_GENERATE_STATIC(_ep_sock_data_tree,
_ep_sock_data,
tree_entry,
_ep_compare_sock_data)
static int _ep_get_related_sockets(_ep_port_data_t* port_data,
SOCKET socket,
SOCKET* afd_socket_out,
SOCKET* driver_socket_out) {
SOCKET afd_socket, driver_socket;
WSAPROTOCOL_INFOW protocol_info;
DWORD bytes;
int len;
/* Try to obtain a base handle for the socket, so we can bypass LSPs
* that get in the way if we want to talk to the kernel directly. If
* it fails we try if we work with the original socket. Note that on
* windows XP/2k3 this will always fail since they don't support the
* SIO_BASE_HANDLE ioctl.
*/
afd_socket = socket;
WSAIoctl(socket,
SIO_BASE_HANDLE,
NULL,
0,
&afd_socket,
sizeof afd_socket,
&bytes,
NULL,
NULL);
/* Obtain protocol information about the socket. */
len = sizeof protocol_info;
if (getsockopt(afd_socket,
SOL_SOCKET,
SO_PROTOCOL_INFOW,
(char*) &protocol_info,
&len) != 0)
return_error(-1);
driver_socket = _ep_get_driver_socket(port_data, &protocol_info);
if (driver_socket == INVALID_SOCKET)
return -1;
*afd_socket_out = afd_socket;
*driver_socket_out = driver_socket;
return 0;
}
static inline _ep_sock_data_t* _ep_sock_data_alloc(void) {
_ep_sock_data_t* sock_data = malloc(sizeof *sock_data);
if (sock_data == NULL)
return_error(NULL, ERROR_NOT_ENOUGH_MEMORY);
return sock_data;
}
static inline void _ep_sock_data_free(_ep_sock_data_t* sock_data) {
free(sock_data);
}
static _ep_sock_data_t* _ep_sock_data_new(_ep_port_data_t* port_data) {
_ep_sock_data_t* sock_data = _ep_sock_data_alloc();
if (sock_data == NULL)
return NULL;
(void) port_data;
memset(sock_data, 0, sizeof *sock_data);
QUEUE_INIT(&sock_data->queue_entry);
return sock_data;
}
void _ep_sock_data_add_poll_req(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data) {
assert(!(sock_data->flags & _EP_SOCK_DELETED));
sock_data->poll_req_count++;
port_data->poll_req_count++;
}
void _ep_sock_data_del_poll_req(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data) {
sock_data->poll_req_count--;
port_data->poll_req_count--;
if ((sock_data->flags & _EP_SOCK_DELETED) && sock_data->poll_req_count == 0)
_ep_sock_data_delete(port_data, sock_data);
}
static int _ep_sock_data_set_socket(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data,
SOCKET socket) {
if (socket == 0 || socket == INVALID_SOCKET)
return_error(-1, ERROR_INVALID_HANDLE);
if (sock_data->socket != 0)
return_error(-1, ERROR_INVALID_PARAMETER);
if (_ep_get_related_sockets(port_data,
socket,
&sock_data->afd_socket,
&sock_data->driver_socket) < 0)
return -1;
sock_data->socket = socket;
if (RB_INSERT(_ep_sock_data_tree, &port_data->sock_data_tree, sock_data) !=
NULL)
return_error(-1, ERROR_ALREADY_EXISTS); /* Socket already in epoll set. */
return 0;
}
static int _ep_sock_data_delete(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data) {
/* Remove from lookup tree. */
if (sock_data->socket != 0 && !(sock_data->flags & _EP_SOCK_DELETED)) {
RB_REMOVE(_ep_sock_data_tree, &port_data->sock_data_tree, sock_data);
sock_data->flags |= _EP_SOCK_DELETED;
}
/* Remove from attention list. */
if (sock_data->flags & _EP_SOCK_LISTED) {
QUEUE_REMOVE(&sock_data->queue_entry);
sock_data->flags &= ~_EP_SOCK_LISTED;
}
/* The socket may still have pending overlapped requests that have yet to be
* reported by the completion port. If that's the case the structure can't be
* free'd yet; epoll_wait() will do it as it dequeues those requests.
*/
if (sock_data->poll_req_count == 0)
_ep_sock_data_free(sock_data);
return 0;
}
static bool _ep_sock_data_marked_for_deletion(_ep_sock_data_t* sock_data) {
return sock_data->flags & _EP_SOCK_DELETED;
}
static int _ep_sock_data_set_event(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data,
const struct epoll_event* ev) {
/* EPOLLERR and EPOLLHUP are always reported, even when no sollicited. */
uint32_t events = ev->events | EPOLLERR | EPOLLHUP;
sock_data->user_events = events;
sock_data->user_data = ev->data;
if (events & _EP_EVENT_MASK & ~(sock_data->latest_poll_req_events)) {
/* Add to attention list, if not already added. */
if (!(sock_data->flags & _EP_SOCK_LISTED)) {
QUEUE_INSERT_TAIL(&port_data->update_queue, &sock_data->queue_entry);
sock_data->flags |= _EP_SOCK_LISTED;
}
}
return 0;
}
static int _ep_ctl_add(_ep_port_data_t* port_data,
uintptr_t socket,
struct epoll_event* ev) {
_ep_sock_data_t* sock_data = _ep_sock_data_new(port_data);
if (sock_data == NULL)
ep_sock_t* sock_info = ep_sock_new(port_data);
if (sock_info == NULL)
return -1;
if (_ep_sock_data_set_socket(port_data, sock_data, socket) < 0 ||
_ep_sock_data_set_event(port_data, sock_data, ev) < 0) {
_ep_sock_data_delete(port_data, sock_data);
if (ep_sock_set_socket(port_data, sock_info, socket) < 0 ||
ep_sock_set_event(port_data, sock_info, ev) < 0) {
ep_sock_delete(port_data, sock_info);
return -1;
}
@ -255,30 +50,30 @@ static int _ep_ctl_add(_ep_port_data_t* port_data,
static int _ep_ctl_mod(_ep_port_data_t* port_data,
uintptr_t socket,
struct epoll_event* ev) {
_ep_sock_data_t lookup;
_ep_sock_data_t* sock_data;
ep_sock_t lookup;
ep_sock_t* sock_info;
lookup.socket = socket;
sock_data = RB_FIND(_ep_sock_data_tree, &port_data->sock_data_tree, &lookup);
if (sock_data == NULL)
sock_info = RB_FIND(ep_sock_tree, &port_data->sock_data_tree, &lookup);
if (sock_info == NULL)
return_error(-1, ERROR_BAD_NETPATH); /* Socket not in epoll set. */
if (_ep_sock_data_set_event(port_data, sock_data, ev) < 0)
if (ep_sock_set_event(port_data, sock_info, ev) < 0)
return -1;
return 0;
}
static int _ep_ctl_del(_ep_port_data_t* port_data, uintptr_t socket) {
_ep_sock_data_t lookup;
_ep_sock_data_t* sock_data;
ep_sock_t lookup;
ep_sock_t* sock_info;
lookup.socket = socket;
sock_data = RB_FIND(_ep_sock_data_tree, &port_data->sock_data_tree, &lookup);
if (sock_data == NULL)
sock_info = RB_FIND(ep_sock_tree, &port_data->sock_data_tree, &lookup);
if (sock_info == NULL)
return_error(-1, ERROR_NOT_FOUND); /* Socket not in epoll set. */
if (_ep_sock_data_delete(port_data, sock_data) < 0)
if (ep_sock_delete(port_data, sock_info) < 0)
return -1;
return 0;
@ -302,42 +97,6 @@ int epoll_ctl(epoll_t port_handle,
return_error(-1, ERROR_INVALID_PARAMETER);
}
static int _ep_sock_update(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data) {
bool broken = false;
assert(sock_data->flags & _EP_SOCK_LISTED);
/* Check if there are events registered that are not yet submitted. In
* that case we need to submit another req.
*/
if ((sock_data->user_events & _EP_EVENT_MASK &
~sock_data->latest_poll_req_events) == 0)
/* All the events the user is interested in are already being monitored
* by the latest poll request. */
goto done;
if (_ep_submit_poll_req(port_data, sock_data) < 0) {
if (GetLastError() == ERROR_INVALID_HANDLE)
/* The socket is broken. It will be dropped from the epoll set. */
broken = true;
else
/* Another error occurred, which is propagated to the caller. */
return -1;
}
done:
/* Remove the socket from the update queue. */
QUEUE_REMOVE(&sock_data->queue_entry);
sock_data->flags &= ~_EP_SOCK_LISTED;
/* If we saw an ERROR_INVALID_HANDLE error, drop the socket.. */
if (broken)
_ep_sock_data_delete(port_data, sock_data);
return 0;
}
static int _ep_port_update_events(_ep_port_data_t* port_data) {
QUEUE* update_queue = &port_data->update_queue;
@ -345,86 +104,18 @@ static int _ep_port_update_events(_ep_port_data_t* port_data) {
* it. */
while (!QUEUE_EMPTY(update_queue)) {
QUEUE* queue_entry = QUEUE_HEAD(update_queue);
_ep_sock_data_t* sock_data =
QUEUE_DATA(queue_entry, _ep_sock_data_t, queue_entry);
ep_sock_t* sock_info = QUEUE_DATA(queue_entry, ep_sock_t, queue_entry);
if (_ep_sock_update(port_data, sock_data) < 0)
if (ep_sock_update(port_data, sock_info) < 0)
return -1;
/* _ep_sock_update() removes the socket from the update list if
/* ep_sock_update() removes the socket from the update list if
* successfull. */
}
return 0;
}
static inline bool _poll_req_is_most_recent(_ep_sock_data_t* sock_data,
poll_req_t* poll_req) {
return poll_req == sock_data->latest_poll_req;
}
static inline void _poll_req_clear_recent(_ep_sock_data_t* sock_data) {
sock_data->latest_poll_req = NULL;
sock_data->latest_poll_req_events = 0;
}
static inline void _poll_req_set_recent(_ep_sock_data_t* sock_data,
poll_req_t* poll_req,
uint32_t epoll_events) {
sock_data->latest_poll_req = poll_req;
sock_data->latest_poll_req_events = epoll_events;
}
static size_t _ep_port_feed_event(_ep_port_data_t* port_data,
poll_req_t* poll_req,
struct epoll_event* ev) {
_ep_sock_data_t* sock_data = poll_req_get_sock_data(poll_req);
uint32_t epoll_events;
bool drop_socket;
size_t ev_count = 0;
if (_ep_sock_data_marked_for_deletion(sock_data) ||
!_poll_req_is_most_recent(sock_data, poll_req)) {
/* Ignore completion for overlapped poll operation if it isn't
* the the most recently posted one, or if the socket has been
* deleted. */
poll_req_delete(port_data, sock_data, poll_req);
return 0;
}
_poll_req_clear_recent(sock_data);
poll_req_complete(poll_req, &epoll_events, &drop_socket);
/* Filter events that the user didn't ask for. */
epoll_events &= sock_data->user_events;
/* Drop the socket if the EPOLLONESHOT flag is set and there are any events
* to report. */
if (epoll_events != 0 && (sock_data->user_events & EPOLLONESHOT))
drop_socket = true;
/* Fill the ev structure if there are any events to report. */
if (epoll_events != 0) {
ev->data = sock_data->user_data;
ev->events = epoll_events;
ev_count = 1;
}
poll_req_delete(port_data, sock_data, poll_req);
if (drop_socket)
/* Drop the socket from the epoll set. */
_ep_sock_data_delete(port_data, sock_data);
else
/* Put the socket back onto the attention list so a new poll request will
* be submitted. */
ATTN_LIST_ADD(port_data, sock_data);
return ev_count;
}
static size_t _ep_port_feed_events(_ep_port_data_t* port_data,
OVERLAPPED_ENTRY* completion_list,
size_t completion_count,
@ -440,7 +131,7 @@ static size_t _ep_port_feed_events(_ep_port_data_t* port_data,
poll_req_t* poll_req = overlapped_to_poll_req(overlapped);
struct epoll_event* ev = &event_list[event_count];
event_count += _ep_port_feed_event(port_data, poll_req, ev);
event_count += ep_sock_feed_event(port_data, poll_req, ev);
}
return event_count;
@ -546,7 +237,7 @@ epoll_t epoll_create(void) {
int epoll_close(epoll_t port_handle) {
_ep_port_data_t* port_data;
_ep_sock_data_t* sock_data;
ep_sock_t* sock_info;
port_data = (_ep_port_data_t*) port_handle;
@ -588,9 +279,9 @@ int epoll_close(epoll_t port_handle) {
}
/* Remove all entries from the socket_state tree. */
while ((sock_data = RB_ROOT(&port_data->sock_data_tree))) {
RB_REMOVE(_ep_sock_data_tree, &port_data->sock_data_tree, sock_data);
free(sock_data);
while ((sock_info = RB_ROOT(&port_data->sock_data_tree))) {
RB_REMOVE(ep_sock_tree, &port_data->sock_data_tree, sock_info);
free(sock_info);
}
/* Close the I/O completion port. */
@ -617,18 +308,29 @@ static int _ep_initialize(void) {
return 0;
}
static SOCKET _ep_get_driver_socket(_ep_port_data_t* port_data,
WSAPROTOCOL_INFOW* protocol_info) {
SOCKET _ep_get_driver_socket(_ep_port_data_t* port_data, SOCKET socket) {
ssize_t index;
size_t i;
SOCKET driver_socket;
WSAPROTOCOL_INFOW protocol_info;
int len;
/* Obtain protocol information about the socket. */
len = sizeof protocol_info;
if (getsockopt(socket,
SOL_SOCKET,
SO_PROTOCOL_INFOW,
(char*) &protocol_info,
&len) != 0)
return_error(INVALID_SOCKET);
index = -1;
for (i = 0; i < array_count(AFD_PROVIDER_GUID_LIST); i++) {
if (memcmp((void*) &protocol_info->ProviderId,
if (memcmp((void*) &protocol_info.ProviderId,
(void*) &AFD_PROVIDER_GUID_LIST[i],
sizeof protocol_info->ProviderId) == 0) {
sizeof protocol_info.ProviderId) == 0) {
index = i;
break;
}
}
@ -642,7 +344,7 @@ static SOCKET _ep_get_driver_socket(_ep_port_data_t* port_data,
*/
driver_socket = port_data->driver_sockets[index];
if (driver_socket == 0) {
driver_socket = _ep_create_driver_socket(port_data->iocp, protocol_info);
driver_socket = _ep_create_driver_socket(port_data->iocp, &protocol_info);
port_data->driver_sockets[index] = driver_socket;
}
@ -675,29 +377,3 @@ error:;
closesocket(socket);
return_error(INVALID_SOCKET, error);
}
int _ep_compare_sock_data(_ep_sock_data_t* a, _ep_sock_data_t* b) {
return a->socket - b->socket;
}
static int _ep_submit_poll_req(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data) {
poll_req_t* poll_req;
uint32_t epoll_events = sock_data->user_events;
poll_req = poll_req_new(port_data, sock_data);
if (poll_req == NULL)
return -1;
if (poll_req_submit(poll_req,
epoll_events,
sock_data->afd_socket,
sock_data->driver_socket) < 0) {
poll_req_delete(port_data, sock_data, poll_req);
return -1;
}
_poll_req_set_recent(sock_data, poll_req, epoll_events);
return 0;
}

19
src/poll-request.c
View File

@ -4,17 +4,17 @@
#include <stdint.h>
#include "afd.h"
#include "epoll-socket.h"
#include "epoll.h"
#include "error.h"
#include "poll-request.h"
#include "sock-data.h"
#include "util.h"
#include "win.h"
typedef struct poll_req {
OVERLAPPED overlapped;
AFD_POLL_INFO poll_info;
_ep_sock_data_t* sock_data;
ep_sock_t* sock_info;
} poll_req_t;
static inline poll_req_t* _poll_req_alloc(void) {
@ -29,26 +29,25 @@ static inline poll_req_t* _poll_req_free(poll_req_t* poll_req) {
return NULL;
}
poll_req_t* poll_req_new(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data) {
poll_req_t* poll_req_new(_ep_port_data_t* port_data, ep_sock_t* sock_info) {
poll_req_t* poll_req = _poll_req_alloc();
if (poll_req == NULL)
return NULL;
memset(poll_req, 0, sizeof *poll_req);
poll_req->sock_data = sock_data;
poll_req->sock_info = sock_info;
_ep_sock_data_add_poll_req(port_data, sock_data);
ep_sock_register_poll_req(port_data, sock_info);
return poll_req;
}
void poll_req_delete(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data,
ep_sock_t* sock_info,
poll_req_t* poll_req) {
assert(poll_req != NULL);
_ep_sock_data_del_poll_req(port_data, sock_data);
ep_sock_unregister_poll_req(port_data, sock_info);
_poll_req_free(poll_req);
}
@ -57,8 +56,8 @@ poll_req_t* overlapped_to_poll_req(OVERLAPPED* overlapped) {
return container_of(overlapped, poll_req_t, overlapped);
}
_ep_sock_data_t* poll_req_get_sock_data(const poll_req_t* poll_req) {
return poll_req->sock_data;
ep_sock_t* poll_req_get_sock_data(const poll_req_t* poll_req) {
return poll_req->sock_info;
}
static DWORD _epoll_events_to_afd_events(uint32_t epoll_events) {

13
src/poll-request.h
View File

@ -1,26 +1,25 @@
#ifndef EPOLL_POLL_REQUEST_H_
#define EPOLL_POLL_REQUEST_H_
#include <stdint.h>
#include <stdbool.h>
#include <stdint.h>
#include "afd.h"
#include "port-data.h"
#include "sock-data.h"
#include "win.h"
typedef struct _ep_port_data _ep_port_data_t;
typedef struct ep_sock ep_sock_t;
typedef struct poll_req poll_req_t;
poll_req_t* poll_req_new(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data);
poll_req_t* poll_req_new(_ep_port_data_t* port_data, ep_sock_t* sock_info);
void poll_req_delete(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data,
ep_sock_t* sock_info,
poll_req_t* poll_req);
poll_req_t* overlapped_to_poll_req(OVERLAPPED* overlapped);
_ep_sock_data_t* poll_req_get_sock_data(const poll_req_t* poll_req);
ep_sock_t* poll_req_get_sock_data(const poll_req_t* poll_req);
int poll_req_submit(poll_req_t* poll_req,
uint32_t epoll_events,

18
src/port-data.h
View File

@ -1,6 +1,24 @@
#ifndef EPOLL_PORT_DATA_H_
#define EPOLL_PORT_DATA_H_
#include "afd.h"
#include "epoll-socket.h"
#include "queue.h"
#include "tree.h"
#include "util.h"
#include "win.h"
typedef struct _ep_port_data _ep_port_data_t;
typedef struct ep_sock ep_sock_t;
typedef struct _ep_port_data {
HANDLE iocp;
SOCKET driver_sockets[array_count(AFD_PROVIDER_GUID_LIST)];
ep_sock_tree_head_t sock_data_tree;
QUEUE update_queue;
size_t poll_req_count;
} _ep_port_data_t;
SOCKET _ep_get_driver_socket(_ep_port_data_t* port_data, SOCKET socket);
#endif /* EPOLL_PORT_DATA_H_ */

13
src/sock-data.h
View File

@ -1,13 +0,0 @@
#ifndef EPOLL_SOCK_DATA_H_
#define EPOLL_SOCK_DATA_H_
#include "port-data.h"
typedef struct _ep_sock_data _ep_sock_data_t;
void _ep_sock_data_add_poll_req(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data);
void _ep_sock_data_del_poll_req(_ep_port_data_t* port_data,
_ep_sock_data_t* sock_data);
#endif /* EPOLL_SOCK_DATA_H_ */