]> git.saurik.com Git - redis.git/blame - ae.c
zero length bulk data reading fixed in loadAppendOnlyFile()
[redis.git] / ae.c
CommitLineData
ed9b544e 1/* A simple event-driven programming library. Originally I wrote this code
2 * for the Jim's event-loop (Jim is a Tcl interpreter) but later translated
3 * it in form of a library for easy reuse.
4 *
5 * Copyright (c) 2006-2009, Salvatore Sanfilippo <antirez at gmail dot com>
6 * All rights reserved.
7 *
8 * Redistribution and use in source and binary forms, with or without
9 * modification, are permitted provided that the following conditions are met:
10 *
11 * * Redistributions of source code must retain the above copyright notice,
12 * this list of conditions and the following disclaimer.
13 * * Redistributions in binary form must reproduce the above copyright
14 * notice, this list of conditions and the following disclaimer in the
15 * documentation and/or other materials provided with the distribution.
16 * * Neither the name of Redis nor the names of its contributors may be used
17 * to endorse or promote products derived from this software without
18 * specific prior written permission.
19 *
20 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
21 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
22 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
23 * ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT OWNER OR CONTRIBUTORS BE
24 * LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
25 * CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
26 * SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
27 * INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
28 * CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
29 * ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
30 * POSSIBILITY OF SUCH DAMAGE.
31 */
32
33#include <stdio.h>
34#include <sys/time.h>
35#include <sys/types.h>
36#include <unistd.h>
37#include <stdlib.h>
38
39#include "ae.h"
40#include "zmalloc.h"
266373b2 41#include "config.h"
42
43/* Include the best multiplexing layer supported by this system.
44 * The following should be ordered by performances, descending. */
45#ifdef HAVE_EPOLL
46#include "ae_epoll.c"
47#else
48#include "ae_select.c"
49#endif
ed9b544e 50
51aeEventLoop *aeCreateEventLoop(void) {
52 aeEventLoop *eventLoop;
266373b2 53 int i;
ed9b544e 54
55 eventLoop = zmalloc(sizeof(*eventLoop));
56 if (!eventLoop) return NULL;
ed9b544e 57 eventLoop->timeEventHead = NULL;
58 eventLoop->timeEventNextId = 0;
59 eventLoop->stop = 0;
266373b2 60 eventLoop->maxfd = -1;
61 if (aeApiCreate(eventLoop) == -1) {
62 zfree(eventLoop);
63 return NULL;
64 }
65 /* Events with mask == AE_NONE are not set. So let's initialize the
66 * vector with it. */
67 for (i = 0; i < AE_SETSIZE; i++)
68 eventLoop->events[i].mask = AE_NONE;
ed9b544e 69 return eventLoop;
70}
71
72void aeDeleteEventLoop(aeEventLoop *eventLoop) {
266373b2 73 aeApiFree(eventLoop);
ed9b544e 74 zfree(eventLoop);
75}
76
77void aeStop(aeEventLoop *eventLoop) {
78 eventLoop->stop = 1;
79}
80
81int aeCreateFileEvent(aeEventLoop *eventLoop, int fd, int mask,
266373b2 82 aeFileProc *proc, void *clientData)
ed9b544e 83{
266373b2 84 if (fd >= AE_SETSIZE) return AE_ERR;
85 aeFileEvent *fe = &eventLoop->events[fd];
86
87 if (aeApiAddEvent(eventLoop, fd, mask) == -1)
88 return AE_ERR;
89 fe->mask |= mask;
90 if (mask & AE_READABLE) fe->rfileProc = proc;
91 if (mask & AE_WRITABLE) fe->wfileProc = proc;
92 if (mask & AE_EXCEPTION) fe->efileProc = proc;
ed9b544e 93 fe->clientData = clientData;
266373b2 94 if (fd > eventLoop->maxfd)
95 eventLoop->maxfd = fd;
ed9b544e 96 return AE_OK;
97}
98
99void aeDeleteFileEvent(aeEventLoop *eventLoop, int fd, int mask)
100{
266373b2 101 if (fd >= AE_SETSIZE) return;
102 aeFileEvent *fe = &eventLoop->events[fd];
103
104 if (fe->mask == AE_NONE) return;
105 fe->mask = fe->mask & (~mask);
106 if (fd == eventLoop->maxfd && fe->mask == AE_NONE) {
107 /* Update the max fd */
108 int j;
109
110 for (j = eventLoop->maxfd-1; j >= 0; j--)
111 if (eventLoop->events[j].mask != AE_NONE) break;
112 eventLoop->maxfd = j;
ed9b544e 113 }
266373b2 114 aeApiDelEvent(eventLoop, fd, mask);
ed9b544e 115}
116
117static void aeGetTime(long *seconds, long *milliseconds)
118{
119 struct timeval tv;
120
121 gettimeofday(&tv, NULL);
122 *seconds = tv.tv_sec;
123 *milliseconds = tv.tv_usec/1000;
124}
125
126static void aeAddMillisecondsToNow(long long milliseconds, long *sec, long *ms) {
127 long cur_sec, cur_ms, when_sec, when_ms;
128
129 aeGetTime(&cur_sec, &cur_ms);
130 when_sec = cur_sec + milliseconds/1000;
131 when_ms = cur_ms + milliseconds%1000;
132 if (when_ms >= 1000) {
133 when_sec ++;
134 when_ms -= 1000;
135 }
136 *sec = when_sec;
137 *ms = when_ms;
138}
139
140long long aeCreateTimeEvent(aeEventLoop *eventLoop, long long milliseconds,
141 aeTimeProc *proc, void *clientData,
142 aeEventFinalizerProc *finalizerProc)
143{
144 long long id = eventLoop->timeEventNextId++;
145 aeTimeEvent *te;
146
147 te = zmalloc(sizeof(*te));
148 if (te == NULL) return AE_ERR;
149 te->id = id;
150 aeAddMillisecondsToNow(milliseconds,&te->when_sec,&te->when_ms);
151 te->timeProc = proc;
152 te->finalizerProc = finalizerProc;
153 te->clientData = clientData;
154 te->next = eventLoop->timeEventHead;
155 eventLoop->timeEventHead = te;
156 return id;
157}
158
159int aeDeleteTimeEvent(aeEventLoop *eventLoop, long long id)
160{
161 aeTimeEvent *te, *prev = NULL;
162
163 te = eventLoop->timeEventHead;
164 while(te) {
165 if (te->id == id) {
166 if (prev == NULL)
167 eventLoop->timeEventHead = te->next;
168 else
169 prev->next = te->next;
170 if (te->finalizerProc)
171 te->finalizerProc(eventLoop, te->clientData);
172 zfree(te);
173 return AE_OK;
174 }
175 prev = te;
176 te = te->next;
177 }
178 return AE_ERR; /* NO event with the specified ID found */
179}
180
181/* Search the first timer to fire.
182 * This operation is useful to know how many time the select can be
183 * put in sleep without to delay any event.
184 * If there are no timers NULL is returned.
185 *
5b2a1c29 186 * Note that's O(N) since time events are unsorted.
187 * Possible optimizations (not needed by Redis so far, but...):
188 * 1) Insert the event in order, so that the nearest is just the head.
189 * Much better but still insertion or deletion of timers is O(N).
190 * 2) Use a skiplist to have this operation as O(1) and insertion as O(log(N)).
191 */
ed9b544e 192static aeTimeEvent *aeSearchNearestTimer(aeEventLoop *eventLoop)
193{
194 aeTimeEvent *te = eventLoop->timeEventHead;
195 aeTimeEvent *nearest = NULL;
196
197 while(te) {
198 if (!nearest || te->when_sec < nearest->when_sec ||
199 (te->when_sec == nearest->when_sec &&
200 te->when_ms < nearest->when_ms))
201 nearest = te;
202 te = te->next;
203 }
204 return nearest;
205}
206
5b2a1c29 207/* Process time events */
208static int processTimeEvents(aeEventLoop *eventLoop) {
209 int processed = 0;
210 aeTimeEvent *te;
211 long long maxId;
212
213 te = eventLoop->timeEventHead;
214 maxId = eventLoop->timeEventNextId-1;
215 while(te) {
216 long now_sec, now_ms;
217 long long id;
218
219 if (te->id > maxId) {
220 te = te->next;
221 continue;
222 }
223 aeGetTime(&now_sec, &now_ms);
224 if (now_sec > te->when_sec ||
225 (now_sec == te->when_sec && now_ms >= te->when_ms))
226 {
227 int retval;
228
229 id = te->id;
230 retval = te->timeProc(eventLoop, id, te->clientData);
231 processed++;
232 /* After an event is processed our time event list may
233 * no longer be the same, so we restart from head.
234 * Still we make sure to don't process events registered
235 * by event handlers itself in order to don't loop forever.
236 * To do so we saved the max ID we want to handle.
237 *
238 * FUTURE OPTIMIZATIONS:
239 * Note that this is NOT great algorithmically. Redis uses
240 * a single time event so it's not a problem but the right
241 * way to do this is to add the new elements on head, and
242 * to flag deleted elements in a special way for later
243 * deletion (putting references to the nodes to delete into
244 * another linked list). */
245 if (retval != AE_NOMORE) {
246 aeAddMillisecondsToNow(retval,&te->when_sec,&te->when_ms);
247 } else {
248 aeDeleteTimeEvent(eventLoop, id);
249 }
250 te = eventLoop->timeEventHead;
251 } else {
252 te = te->next;
253 }
254 }
255 return processed;
256}
257
ed9b544e 258/* Process every pending time event, then every pending file event
259 * (that may be registered by time event callbacks just processed).
260 * Without special flags the function sleeps until some file event
261 * fires, or when the next time event occurrs (if any).
262 *
263 * If flags is 0, the function does nothing and returns.
264 * if flags has AE_ALL_EVENTS set, all the kind of events are processed.
265 * if flags has AE_FILE_EVENTS set, file events are processed.
266 * if flags has AE_TIME_EVENTS set, time events are processed.
267 * if flags has AE_DONT_WAIT set the function returns ASAP until all
268 * the events that's possible to process without to wait are processed.
269 *
270 * The function returns the number of events processed. */
271int aeProcessEvents(aeEventLoop *eventLoop, int flags)
272{
266373b2 273 int processed = 0, numevents;
ed9b544e 274
275 /* Nothing to do? return ASAP */
276 if (!(flags & AE_TIME_EVENTS) && !(flags & AE_FILE_EVENTS)) return 0;
277
ed9b544e 278 /* Note that we want call select() even if there are no
279 * file events to process as long as we want to process time
280 * events, in order to sleep until the next time event is ready
281 * to fire. */
266373b2 282 if (eventLoop->maxfd != -1 ||
283 ((flags & AE_TIME_EVENTS) && !(flags & AE_DONT_WAIT))) {
284 int j;
ed9b544e 285 aeTimeEvent *shortest = NULL;
286 struct timeval tv, *tvp;
287
288 if (flags & AE_TIME_EVENTS && !(flags & AE_DONT_WAIT))
289 shortest = aeSearchNearestTimer(eventLoop);
290 if (shortest) {
291 long now_sec, now_ms;
292
293 /* Calculate the time missing for the nearest
294 * timer to fire. */
295 aeGetTime(&now_sec, &now_ms);
296 tvp = &tv;
297 tvp->tv_sec = shortest->when_sec - now_sec;
298 if (shortest->when_ms < now_ms) {
299 tvp->tv_usec = ((shortest->when_ms+1000) - now_ms)*1000;
300 tvp->tv_sec --;
301 } else {
302 tvp->tv_usec = (shortest->when_ms - now_ms)*1000;
303 }
266373b2 304 if (tvp->tv_sec < 0) tvp->tv_sec = 0;
305 if (tvp->tv_usec < 0) tvp->tv_usec = 0;
ed9b544e 306 } else {
307 /* If we have to check for events but need to return
308 * ASAP because of AE_DONT_WAIT we need to se the timeout
309 * to zero */
310 if (flags & AE_DONT_WAIT) {
311 tv.tv_sec = tv.tv_usec = 0;
312 tvp = &tv;
313 } else {
314 /* Otherwise we can block */
315 tvp = NULL; /* wait forever */
316 }
317 }
318
266373b2 319 numevents = aeApiPoll(eventLoop, tvp);
320 for (j = 0; j < numevents; j++) {
321 aeFileEvent *fe = &eventLoop->events[eventLoop->fired[j].fd];
322 int mask = eventLoop->fired[j].mask;
323 int fd = eventLoop->fired[j].fd;
324
325 /* note the fe->mask & mask & ... code: maybe an already processed
326 * event removed an element that fired and we still didn't
327 * processed, so we check if the event is still valid. */
328 if (fe->mask & mask & AE_READABLE)
329 fe->rfileProc(eventLoop,fd,fe->clientData,mask);
330 if (fe->mask & mask & AE_WRITABLE && fe->wfileProc != fe->rfileProc)
331 fe->wfileProc(eventLoop,fd,fe->clientData,mask);
332 if (fe->mask & mask & AE_EXCEPTION &&
333 fe->efileProc != fe->wfileProc &&
334 fe->efileProc != fe->rfileProc)
335 fe->efileProc(eventLoop,fd,fe->clientData,mask);
336 processed++;
ed9b544e 337 }
338 }
339 /* Check time events */
5b2a1c29 340 if (flags & AE_TIME_EVENTS)
341 processed += processTimeEvents(eventLoop);
ed9b544e 342
ed9b544e 343 return processed; /* return the number of processed file/time events */
344}
345
346/* Wait for millseconds until the given file descriptor becomes
347 * writable/readable/exception */
348int aeWait(int fd, int mask, long long milliseconds) {
349 struct timeval tv;
350 fd_set rfds, wfds, efds;
351 int retmask = 0, retval;
352
353 tv.tv_sec = milliseconds/1000;
354 tv.tv_usec = (milliseconds%1000)*1000;
355 FD_ZERO(&rfds);
356 FD_ZERO(&wfds);
357 FD_ZERO(&efds);
358
359 if (mask & AE_READABLE) FD_SET(fd,&rfds);
360 if (mask & AE_WRITABLE) FD_SET(fd,&wfds);
361 if (mask & AE_EXCEPTION) FD_SET(fd,&efds);
362 if ((retval = select(fd+1, &rfds, &wfds, &efds, &tv)) > 0) {
363 if (FD_ISSET(fd,&rfds)) retmask |= AE_READABLE;
364 if (FD_ISSET(fd,&wfds)) retmask |= AE_WRITABLE;
365 if (FD_ISSET(fd,&efds)) retmask |= AE_EXCEPTION;
366 return retmask;
367 } else {
368 return retval;
369 }
370}
371
372void aeMain(aeEventLoop *eventLoop)
373{
374 eventLoop->stop = 0;
375 while (!eventLoop->stop)
376 aeProcessEvents(eventLoop, AE_ALL_EVENTS);
377}