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
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
|
/*
* Copyright (c) 2009-2011, Salvatore Sanfilippo <antirez at gmail dot com>
*
* SPDX-FileCopyrightText: 2024 Hiredict Contributors
* SPDX-FileCopyrightText: 2024 Salvatore Sanfilippo <antirez at gmail dot com>
*
* SPDX-License-Identifier: BSD-3-Clause
* SPDX-License-Identifier: LGPL-3.0-or-later
*
*/
#ifndef HIREDICT_POLL_H
#define HIREDICT_POLL_H
#include "../async.h"
#include "../sockcompat.h"
#include <string.h> // for memset
#include <errno.h>
/* Values to return from redictPollTick */
#define REDICT_POLL_HANDLED_READ 1
#define REDICT_POLL_HANDLED_WRITE 2
#define REDICT_POLL_HANDLED_TIMEOUT 4
/* An adapter to allow manual polling of the async context by checking the state
* of the underlying file descriptor. Useful in cases where there is no formal
* IO event loop but regular ticking can be used, such as in game engines. */
typedef struct redictPollEvents {
redictAsyncContext *context;
redictFD fd;
char reading, writing;
char in_tick;
char deleted;
double deadline;
} redictPollEvents;
static double redictPollTimevalToDouble(struct timeval *tv) {
if (tv == NULL)
return 0.0;
return tv->tv_sec + tv->tv_usec / 1000000.00;
}
static double redictPollGetNow(void) {
#ifndef _MSC_VER
struct timeval tv;
gettimeofday(&tv,NULL);
return redictPollTimevalToDouble(&tv);
#else
FILETIME ft;
ULARGE_INTEGER li;
GetSystemTimeAsFileTime(&ft);
li.HighPart = ft.dwHighDateTime;
li.LowPart = ft.dwLowDateTime;
return (double)li.QuadPart * 1e-7;
#endif
}
/* Poll for io, handling any pending callbacks. The timeout argument can be
* positive to wait for a maximum given time for IO, zero to poll, or negative
* to wait forever */
static int redictPollTick(redictAsyncContext *ac, double timeout) {
int reading, writing;
struct pollfd pfd;
int handled;
int ns;
int itimeout;
redictPollEvents *e = (redictPollEvents*)ac->ev.data;
if (!e)
return 0;
/* local flags, won't get changed during callbacks */
reading = e->reading;
writing = e->writing;
if (!reading && !writing)
return 0;
pfd.fd = e->fd;
pfd.events = 0;
if (reading)
pfd.events = POLLIN;
if (writing)
pfd.events |= POLLOUT;
if (timeout >= 0.0) {
itimeout = (int)(timeout * 1000.0);
} else {
itimeout = -1;
}
ns = poll(&pfd, 1, itimeout);
if (ns < 0) {
/* ignore the EINTR error */
if (errno != EINTR)
return ns;
ns = 0;
}
handled = 0;
e->in_tick = 1;
if (ns) {
if (reading && (pfd.revents & POLLIN)) {
redictAsyncHandleRead(ac);
handled |= REDICT_POLL_HANDLED_READ;
}
/* on Windows, connection failure is indicated with the Exception fdset.
* handle it the same as writable. */
if (writing && (pfd.revents & (POLLOUT | POLLERR))) {
/* context Read callback may have caused context to be deleted, e.g.
by doing an redictAsyncDisconnect() */
if (!e->deleted) {
redictAsyncHandleWrite(ac);
handled |= REDICT_POLL_HANDLED_WRITE;
}
}
}
/* perform timeouts */
if (!e->deleted && e->deadline != 0.0) {
double now = redictPollGetNow();
if (now >= e->deadline) {
/* deadline has passed. disable timeout and perform callback */
e->deadline = 0.0;
redictAsyncHandleTimeout(ac);
handled |= REDICT_POLL_HANDLED_TIMEOUT;
}
}
/* do a delayed cleanup if required */
if (e->deleted)
hi_free(e);
else
e->in_tick = 0;
return handled;
}
static void redictPollAddRead(void *data) {
redictPollEvents *e = (redictPollEvents*)data;
e->reading = 1;
}
static void redictPollDelRead(void *data) {
redictPollEvents *e = (redictPollEvents*)data;
e->reading = 0;
}
static void redictPollAddWrite(void *data) {
redictPollEvents *e = (redictPollEvents*)data;
e->writing = 1;
}
static void redictPollDelWrite(void *data) {
redictPollEvents *e = (redictPollEvents*)data;
e->writing = 0;
}
static void redictPollCleanup(void *data) {
redictPollEvents *e = (redictPollEvents*)data;
/* if we are currently processing a tick, postpone deletion */
if (e->in_tick)
e->deleted = 1;
else
hi_free(e);
}
static void redictPollScheduleTimer(void *data, struct timeval tv)
{
redictPollEvents *e = (redictPollEvents*)data;
double now = redictPollGetNow();
e->deadline = now + redictPollTimevalToDouble(&tv);
}
static int redictPollAttach(redictAsyncContext *ac) {
redictContext *c = &(ac->c);
redictPollEvents *e;
/* Nothing should be attached when something is already attached */
if (ac->ev.data != NULL)
return REDICT_ERR;
/* Create container for context and r/w events */
e = (redictPollEvents*)hi_malloc(sizeof(*e));
if (e == NULL)
return REDICT_ERR;
memset(e, 0, sizeof(*e));
e->context = ac;
e->fd = c->fd;
e->reading = e->writing = 0;
e->in_tick = e->deleted = 0;
e->deadline = 0.0;
/* Register functions to start/stop listening for events */
ac->ev.addRead = redictPollAddRead;
ac->ev.delRead = redictPollDelRead;
ac->ev.addWrite = redictPollAddWrite;
ac->ev.delWrite = redictPollDelWrite;
ac->ev.scheduleTimer = redictPollScheduleTimer;
ac->ev.cleanup = redictPollCleanup;
ac->ev.data = e;
return REDICT_OK;
}
#endif /* HIREDICT_POLL_H */
|
